// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: authentication-free shared key and token derivation between clients
// feature: listener code decoupling
// feature: secure discovery server and client
// feature: support auth token
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: the default memory policy change from PREALLOCATED_MEMORY_MODE to PREALLOCATED_WITH_REALLOC_MEMORY_MODE
// ------------------------------------------------------------------

#include <rtps/security/SecurityManager.h>

#include <elog/Log.hpp>

#include <edds/rtps/attributes/HistoryAttributes.h>
#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <discovery/endpoint/EDP.h>
#include <discovery/participant/PDP.h>
#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>
#include <edds/rtps/messages/CDRMessage.h>
#include <transport/TransportFactory.hpp>
#include <history/ParticipantListener.h>
#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/reader/StatelessReader.h>
#include <edds/rtps/security/accesscontrol/AccessControl.h>
#include <edds/rtps/security/accesscontrol/EndpointSecurityAttributes.h>
#include <edds/rtps/security/accesscontrol/ParticipantSecurityAttributes.h>
#include <edds/rtps/security/accesscontrol/SecurityMaskUtilities.h>
#include <edds/rtps/security/authentication/Authentication.h>
#include <edds/rtps/security/exceptions/SecurityException.h>
#include <edds/rtps/writer/StatefulWriter.h>
#include <edds/rtps/writer/StatelessWriter.h>
// #include <security/authentication/PKIIdentityHandle.h>

#include <security/authentication/PKIDH.h>
#include <security/authentication/PKIIdentityHandle.h>
#include <rtps/participant/RTPSParticipantImpl.h>

#include <cassert>
#include <chrono>
#include <thread>
#include <mutex>

#include <history/TopicPayloadPoolRegistry.hpp>
#include <security/OpenSSLInit.hpp>
#include <string>

#include <edds/rtps/participant/RTPSParticipant.h>
#include <edds/rtps/participant/ParticipantDiscoveryInfo.h>

#define BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_SECURE_WRITER (1 << 20)
#define BUILTIN_ENDPOINT_PARTICIPANT_MESSAGE_SECURE_READER (1 << 21)
#define BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_WRITER (1 << 22)
#define BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_READER (1 << 23)
#define BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER (1 << 24)
#define BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER (1 << 25)

#define AUTHENTICATION_PARTICIPANT_STATELESS_MESSAGE "dds.sec.auth"
#define GMCLASSID_SECURITY_PARTICIPANT_CRYPTO_TOKENS "dds.sec.participant_crypto_tokens"
#define GMCLASSID_SECURITY_READER_CRYPTO_TOKENS "dds.sec.datareader_crypto_tokens"
#define GMCLASSID_SECURITY_WRITER_CRYPTO_TOKENS "dds.sec.datawriter_crypto_tokens"

// TODO(Ricardo) Add event because stateless messages can be not received.

namespace evbs {
namespace ertps {
namespace rtps {

using namespace security;
using namespace vbs::common;

inline bool usleep_bool() {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    return true;
}

static CacheChange_t* create_change_for_message(const ParticipantGenericMessage& message, WriterHistory* history) {
    uint32_t cdr_size = static_cast<uint32_t>(ParticipantGenericMessageHelper::serialized_size(message));
    cdr_size += 4 + 4;  // Encapsulation
    return history->new_change([cdr_size]() -> uint32_t { return cdr_size; }, ALIVE, c_InstanceHandle_Unknown);
}

SecurityManager::SecurityManager(RTPSParticipantImpl* participant)
    : participant_stateless_message_listener_(*this),
      participant_volatile_message_secure_listener_(*this),
      participant_(participant),
      participant_stateless_message_writer_(nullptr),
      participant_stateless_message_writer_history_(nullptr),
      participant_stateless_message_reader_(nullptr),
      participant_stateless_message_reader_history_(nullptr),
      participant_volatile_message_secure_writer_(nullptr),
      participant_volatile_message_secure_writer_history_(nullptr),
      participant_volatile_message_secure_reader_(nullptr),
      participant_volatile_message_secure_reader_history_(nullptr),
      logging_plugin_(nullptr),
      authentication_plugin_(nullptr),
      access_plugin_(nullptr),
      crypto_plugin_(nullptr),
      domain_id_(0),
      local_identity_handle_(nullptr),
      local_permissions_handle_(nullptr),
      auth_last_sequence_number_(1),
      crypto_last_sequence_number_(1),
      temp_reader_proxies_({
          participant->getRTPSParticipantAttributes().allocation.locators.max_unicast_locators,
          participant->getRTPSParticipantAttributes().allocation.locators.max_multicast_locators,
          participant->getRTPSParticipantAttributes().allocation.data_limits,
          // participant->getRTPSParticipantAttributes().allocation.content_filter
      }),
      temp_writer_proxies_({participant->getRTPSParticipantAttributes().allocation.locators.max_unicast_locators,
                            participant->getRTPSParticipantAttributes().allocation.locators.max_multicast_locators,
                            participant->getRTPSParticipantAttributes().allocation.data_limits}) {
    assert(participant != nullptr);
}

SecurityManager::~SecurityManager() {
    destroy();
}

bool SecurityManager::init(ParticipantSecurityAttributes& attributes, const PropertyPolicy& participant_properties) {
    SecurityException exception;
    try {
        domain_id_ = participant_->get_domain_id();
        const auto& part_attributes = participant_->getRTPSParticipantAttributes();
        const PropertyPolicy log_properties = PropertyPolicyHelper::get_properties_with_prefix(
            part_attributes.properties, "dds.sec.log.builtin.DDS_LogTopic.");

        // length(log_properties) == 0 considered as logging disable.
        if (PropertyPolicyHelper::length(log_properties) > 0) {
            logging_plugin_ = factory_.create_logging_plugin(participant_properties);

            if (logging_plugin_ != nullptr) {
                LogOptions log_options;
                log_options.distribute = false;
                log_options.log_level = LoggingLevel::ERROR_LEVEL;
                log_options.log_file = "";

                const std::string* const distribute = PropertyPolicyHelper::find_property(log_properties, "distribute");
                if (distribute != nullptr) {
                    if ((!distribute->compare("true")) || (!distribute->compare("1"))) {
                        log_options.distribute = true;
                    } else if ((!distribute->compare("false")) || (!distribute->compare("0"))) {
                        log_options.distribute = false;
                    } else {
                        throw SecurityException("Unknown value '" + *distribute + "' for LogOptions::distribute.");
                    }
                }

                const std::string* const log_level =
                    PropertyPolicyHelper::find_property(log_properties, "logging_level");
                if (log_level != nullptr) {
                    if (!string_to_LogLevel(*log_level, log_options.log_level, exception)) {
                        throw exception;
                    }
                }

                const std::string* const log_file = PropertyPolicyHelper::find_property(log_properties, "log_file");
                if (log_file != nullptr) {
                    log_options.log_file = *log_file;
                }

                if (!(((logging_plugin_->set_guid(participant_->getGuid(), exception))) &&
                      logging_plugin_->set_domain_id(domain_id_, exception))) {
                    throw exception;
                }

                if (!(((logging_plugin_->set_log_options(log_options, exception))) &&
                      logging_plugin_->enable_logging(exception))) {
                    throw exception;
                }
            } else {
                // TODO(artivis): If the factory fails instantiating 'authentication_plugin_',
                //  a logDebug is issued and this init function returns true. Is it a bug?
                //  in the meantime we'll adopt a similar behavior here.
                logDebug(SECURITY, "Could not create logging plugin. Security logging will be disabled.");
            }
        }

        const auto log_info_message = [this](const char* msg) {
            const char* safe_msg = (msg && msg[0] != '\0') ? msg : "Unknown or empty exception";
            if (logging_plugin_) {
                SecurityException logging_exception;
                logging_plugin_->log(LoggingLevel::INFORMATIONAL_LEVEL, safe_msg, "SecurityManager,init",
                                     logging_exception);
            } else {
                logInfo(SECURITY, safe_msg);
            }
        };

        authentication_plugin_ = factory_.create_authentication_plugin(participant_properties);

        if (authentication_plugin_ != nullptr) {

            // retrieve authentication properties, if any
            const PropertyPolicy auth_handshake_properties = PropertyPolicyHelper::get_properties_with_prefix(
                part_attributes.properties, "dds.sec.auth.builtin.PKI-DH.");

            // if auth_handshake_properties is empty, the default values are used
            if (PropertyPolicyHelper::length(auth_handshake_properties) > 0) {
                auth_handshake_props_.parse_from_property_policy(auth_handshake_properties);
            }

            (void)authentication_plugin_->set_logger(logging_plugin_, exception);

            // Validate local participant
            GUID_t adjusted_participant_key;
            ValidationResult_t ret = VALIDATION_FAILED;

            do {
                ret = authentication_plugin_->validate_local_identity(&local_identity_handle_, adjusted_participant_key,
                                                                      domain_id_, part_attributes,
                                                                      participant_->getGuid(), exception);
            } while ((ret == VALIDATION_PENDING_RETRY) && usleep_bool());

            if (ret == VALIDATION_OK) {
                assert(local_identity_handle_ != nullptr);
                assert(!local_identity_handle_->nil());

                // Set participant guid
                participant_->setGuid(adjusted_participant_key);

                access_plugin_ = factory_.create_access_control_plugin(participant_properties);

                if (access_plugin_ != nullptr) {
                    (void)access_plugin_->set_logger(logging_plugin_, exception);

                    local_permissions_handle_ = access_plugin_->validate_local_permissions(
                        *authentication_plugin_, *local_identity_handle_, domain_id_, part_attributes, exception);

                    if (local_permissions_handle_ != nullptr) {
                        if (!local_permissions_handle_->nil()) {
                            if (access_plugin_->check_create_participant(*local_permissions_handle_, domain_id_,
                                                                         part_attributes, exception)) {
                                // Set credentials.
                                PermissionsCredentialToken* token = nullptr;
                                if (access_plugin_->get_permissions_credential_token(&token, *local_permissions_handle_,
                                                                                     exception)) {
                                    if (authentication_plugin_->set_permissions_credential_and_token(
                                            *local_identity_handle_, *token, exception)) {
                                        if (!access_plugin_->get_participant_sec_attributes(*local_permissions_handle_,
                                                                                            attributes, exception)) {
                                            (void)access_plugin_->return_permissions_handle(local_permissions_handle_,
                                                                                            exception);
                                            local_permissions_handle_ = nullptr;
                                        }
                                    } else {
                                        (void)access_plugin_->return_permissions_handle(local_permissions_handle_,
                                                                                        exception);
                                        local_permissions_handle_ = nullptr;
                                    }

                                    (void)access_plugin_->return_permissions_credential_token(token, exception);
                                } else {
                                    (void)access_plugin_->return_permissions_handle(local_permissions_handle_,
                                                                                    exception);
                                    local_permissions_handle_ = nullptr;
                                }
                            } else {
                                (void)access_plugin_->return_permissions_handle(local_permissions_handle_, exception);
                                local_permissions_handle_ = nullptr;
                            }
                        } else {
                            (void)access_plugin_->return_permissions_handle(local_permissions_handle_, exception);
                            local_permissions_handle_ = nullptr;
                        }
                    } else {
                        log_info_message(exception.what());
                    }
                } else {
                    log_info_message("Access control plugin not configured");
                }

                if ((access_plugin_ == nullptr) || (local_permissions_handle_ != nullptr)) {
                    crypto_plugin_ = factory_.create_cryptography_plugin(participant_properties);

                    if (crypto_plugin_ != nullptr) {
                        (void)crypto_plugin_->set_logger(logging_plugin_, exception);

                        local_participant_crypto_handle_ =
                            crypto_plugin_->cryptokeyfactory()->register_local_participant(
                                *local_identity_handle_, *local_permissions_handle_,
                                participant_properties.properties(), attributes, exception);

                        if (local_participant_crypto_handle_) {
                            assert(!local_participant_crypto_handle_->nil());
                        } else {
                            log_info_message(exception.what());
                        }
                    } else {
                        log_info_message("Cryptography plugin not configured");
                    }
                } else {
                    log_info_message("Validate local permissions failed with access control plugin");
                }

                if (((access_plugin_ == nullptr) || (local_permissions_handle_ != nullptr)) &&
                    ((crypto_plugin_ == nullptr) || local_participant_crypto_handle_)) {
                    // Should be activated here, to enable encription buffer on created entities
                    throw true;
                } else {
                    if (access_plugin_ != nullptr && local_permissions_handle_ == nullptr) {
                        logInfo(SECURITY, "Participant is not allowed with its own permissions file.");
                    }

                    if (crypto_plugin_ != nullptr && local_participant_crypto_handle_ == nullptr) {
                        logInfo(SECURITY, "Participant cryptography could not be configured.");
                    }
                }
            }

            throw false;
        } else {
            logDebug(SECURITY, "Authentication plugin not configured. Security will be disabled");
        }
    } catch (const SecurityException& e) {
        elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED,
                  "Logging plugin not configured. Security logging will be disabled. (" << e.what() << ").");
        delete logging_plugin_;
        logging_plugin_ = nullptr;
        return false;
    } catch (bool e) {
        if (!e) {
            // Unexpected code path. Let's log any errors
            elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED, "Error while configuring security plugin.");
            if (0 != strlen(exception.what())) {
                elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED, exception.what());
            }

            cancel_init();
            return false;
        }
    }

    enable_security_manager();
    return true;
}

void SecurityManager::cancel_init() {  // LCOV_EXCL_START
    SecurityException exception;
    if (local_participant_crypto_handle_) {
        (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(local_participant_crypto_handle_, exception);
    }

    if (crypto_plugin_ != nullptr) {
        delete crypto_plugin_;
        crypto_plugin_ = nullptr;
    }

    // TODO(Ricardo) Return local_permissions

    if (access_plugin_ != nullptr) {
        if (local_permissions_handle_ != nullptr) {
            access_plugin_->return_permissions_handle(local_permissions_handle_, exception);
        }

        delete access_plugin_;
        access_plugin_ = nullptr;
    }

    if (authentication_plugin_ != nullptr) {
        if (local_identity_handle_ != nullptr) {
            authentication_plugin_->return_identity_handle(local_identity_handle_, exception);
        }

        delete authentication_plugin_;
        authentication_plugin_ = nullptr;
    }

    disable_security_manager();
}  // LCOV_EXCL_STOP

void SecurityManager::destroy() {
    disable_security_manager();

    if (authentication_plugin_ != nullptr) {
        vbs::shared_lock<vbs::shared_mutex> _(mutex_);

        SecurityException exception;

        // writer_handles_ and reader_handles_ contents reference the discovered_participants_ ones
        // and are disposed on this collection removal

        for (auto& dp_it : discovered_participants_) {
            std::shared_ptr<ParticipantCryptoHandle> participant_crypto_handle = dp_it.second->get_participant_crypto();
            if (participant_crypto_handle != nullptr) {
                (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(participant_crypto_handle, exception);
            }

            PermissionsHandle* permissions_handle = dp_it.second->get_permissions_handle();
            if (permissions_handle != nullptr) {
                (void)access_plugin_->return_permissions_handle(permissions_handle, exception);
            }

            std::shared_ptr<SecretHandle> shared_secret_handle = dp_it.second->get_shared_secret();
            if (shared_secret_handle != nullptr) {
                (void)authentication_plugin_->return_sharedsecret_handle(shared_secret_handle, exception);
            }

            remove_discovered_participant_info(dp_it.second->get_auth());
        }

        if (local_participant_crypto_handle_) {
            (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(local_participant_crypto_handle_,
                                                                             exception);
        }

        if (local_permissions_handle_ != nullptr) {
            (void)access_plugin_->return_permissions_handle(local_permissions_handle_, exception);
        }

        if (local_identity_handle_ != nullptr) {
            (void)authentication_plugin_->return_identity_handle(local_identity_handle_, exception);
        }
    }

    std::lock_guard<vbs::shared_mutex> _(mutex_);

    // check the logic
    delete_entities();

    writer_handles_.clear();
    reader_handles_.clear();
    discovered_participants_.clear();

    local_permissions_handle_ = nullptr;
    local_identity_handle_ = nullptr;

    if (crypto_plugin_ != nullptr) {
        delete crypto_plugin_;
        crypto_plugin_ = nullptr;
    }

    if (access_plugin_ != nullptr) {
        delete access_plugin_;
        access_plugin_ = nullptr;
    }

    if (authentication_plugin_ != nullptr) {
        delete authentication_plugin_;
        authentication_plugin_ = nullptr;
    }

    if (logging_plugin_ != nullptr) {
        delete logging_plugin_;
        logging_plugin_ = nullptr;
    }
}

void SecurityManager::remove_discovered_participant_info(const DiscoveredParticipantInfo::AuthUniquePtr& auth_ptr) {
    SecurityException exception;

    if (auth_ptr) {
        if (auth_ptr->handshake_handle_ != nullptr) {
            (void)authentication_plugin_->return_handshake_handle(auth_ptr->handshake_handle_, exception);
            auth_ptr->handshake_handle_ = nullptr;
        }

        (void)authentication_plugin_->return_identity_handle(auth_ptr->identity_handle_, exception);
        auth_ptr->identity_handle_ = nullptr;

        if (auth_ptr->change_sequence_number_ != SequenceNumber_t::unknown()) {
            (void)participant_stateless_message_writer_history_->remove_change(auth_ptr->change_sequence_number_);
            auth_ptr->change_sequence_number_ = SequenceNumber_t::unknown();
        }
    }
}

bool SecurityManager::restore_discovered_participant_info(const GUID_t& remote_participant_key,
                                                          DiscoveredParticipantInfo::AuthUniquePtr& auth_ptr) {
    SecurityException exception;
    bool returned_value = false;

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    auto dp_it = discovered_participants_.find(remote_participant_key);

    if (dp_it != discovered_participants_.end()) {
        dp_it->second->set_auth(auth_ptr);
        returned_value = true;
    } else {
        remove_discovered_participant_info(std::move(auth_ptr));
    }

    return returned_value;
}

bool SecurityManager::discovered_participant(const ParticipantProxyData& participant_data) {
    // std::lock_guard<std::mutex> lck(mtx);
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    // Early return when ParticipantSecurityInfo does not match
    auto& sec_attrs = participant_->security_attributes();
    if (!sec_attrs.match(participant_data.security_attributes_, participant_data.plugin_security_attributes_)) {
        logDebug(SECURITY, "participant security_attributes unmatch with plugin: "
                               << participant_data.security_attributes_ << " "
                               << participant_data.plugin_security_attributes_);
        return false;
    }

    if (authentication_plugin_ == nullptr) {
        participant_->pdp()->notifyAboveRemoteEndpoints(participant_data, true);
        return true;
    }

    SecurityException exception;
    AuthenticationStatus auth_status = AUTHENTICATION_INIT;

    // Create or find information
    bool undiscovered = false;
    bool need_send_request = false;
    DiscoveredParticipantInfo::AuthUniquePtr remote_participant_info;

    // Use the information from the collection
    ParticipantProxyData remote_participant_data;
    {
        std::lock_guard<vbs::shared_mutex> _(mutex_);

        auto map_ret = discovered_participants_.insert(std::make_pair(
            participant_data.m_guid,
            std::unique_ptr<DiscoveredParticipantInfo>(new DiscoveredParticipantInfo(auth_status, participant_data))));

        undiscovered = map_ret.second;
        remote_participant_info = map_ret.first->second->get_auth();
        remote_participant_data = map_ret.first->second->participant_data();
        need_send_request = undiscovered;
    }

    // A new remote participant restarts without changing its guid and needs to be re-authentication
    if (!undiscovered && remote_participant_data.m_startTime != 0 &&
        remote_participant_data.m_startTime != participant_data.m_startTime) {
        remove_participant(remote_participant_data);

        std::lock_guard<vbs::shared_mutex> _(mutex_);

        if (remote_participant_info) {
            remove_discovered_participant_info(remote_participant_info);
        }

        auto map_ret = discovered_participants_.insert(std::make_pair(
            participant_data.m_guid,
            std::unique_ptr<DiscoveredParticipantInfo>(new DiscoveredParticipantInfo(auth_status, participant_data))));

        undiscovered = map_ret.second;
        remote_participant_info = map_ret.first->second->get_auth();
        remote_participant_data = map_ret.first->second->participant_data();
        logInfo(SECURITY, "Reauthentication for:" << participant_data.m_guid << " local:" << participant_->getGuid());
    }

    bool notify_part_authorized = false;
    if (undiscovered && remote_participant_info) {
        // Configure the timed event but do not start it
        const GUID_t guid = remote_participant_data.m_guid;

        remote_participant_info->event_.reset(new TimedEvent(
            participant_->getEventResource(),
            [&, guid]() -> bool {
                resend_handshake_message_token(guid);
                return true;
            },
            static_cast<double>(auth_handshake_props_.initial_handshake_resend_period_ms_)));

        IdentityHandle* remote_identity_handle = nullptr;

        ValidationResult_t validation_ret = authentication_plugin_->validate_remote_identity(
            &remote_identity_handle, *local_identity_handle_, remote_participant_data.identity_token_,
            remote_participant_data.m_guid, exception);
        switch (validation_ret) {
            case VALIDATION_OK:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_OK;
                break;
            case VALIDATION_PENDING_HANDSHAKE_REQUEST:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_REQUEST_NOT_SEND;
                break;
            case VALIDATION_PENDING_HANDSHAKE_MESSAGE:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_WAITING_REQUEST;
                break;
            case VALIDATION_PENDING_RETRY:
            // TODO(Ricardo) Send event.
            default:
                on_validation_failed(remote_participant_data, exception);

                std::unique_ptr<DiscoveredParticipantInfo> tmp;
                std::lock_guard<vbs::shared_mutex> _(mutex_);
                {
                    auto it = discovered_participants_.find(remote_participant_data.m_guid);
                    if (it != discovered_participants_.end()) {
                        tmp = std::move(it->second);
                    }

                    // Remove created element, because authentication failed.
                    (void)discovered_participants_.erase(remote_participant_data.m_guid);
                }

                // TODO(Ricardo) cryptograhy registration in AUTHENTICAITON_OK
                return false;
        }

        logInfo(SECURITY, "Discovered participant " << remote_participant_data.m_guid);

        // Match entities
        match_builtin_endpoints(remote_participant_data);

        // Store new remote handle.
        remote_participant_info->auth_status_ = auth_status;
        remote_participant_info->identity_handle_ = remote_identity_handle;

        // TODO(Ricardo) Start cryptography if authentication ok in this point.
        // If authentication is successful, inform user about it.
        if (auth_status == AUTHENTICATION_OK) {
            // TODO(Ricardo) Shared secret on this case?
            std::shared_ptr<SecretHandle> ss;
            notify_part_authorized = participant_authorized(remote_participant_data, remote_participant_info, ss);
        }
    } else {
        // If cannot retrieve the authentication info pointer then return, because
        // it is used in other thread.
        if (!remote_participant_info) {
            return false;
        }
        if (remote_participant_info->auth_status_ == AUTHENTICATION_FAILED) {
            remote_participant_info->auth_status_ = AUTHENTICATION_REQUEST_NOT_SEND;
        }
    }

    // Modify protocol specifications
    if (need_send_request && remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST &&
        participant_->pdp()->getLocalParticipantProxyData()) {
        bool notify = false;
        bool ret = false;
        // Send request
        remote_participant_info->auth_status_ = AUTHENTICATION_REQUEST_NOT_SEND;
        ret = on_process_handshake(remote_participant_data, remote_participant_info, MessageIdentity(),
                                   HandshakeMessageToken(), notify);
        logInfo(SECURITY, "Participant with larger GUID also send request from:" << participant_->getGuid() << " to:"
                                                                                 << remote_participant_data.m_guid
                                                                                 << " success:" << ret);
        if (ret) {
            // Remove previous change
            remote_participant_info->event_->cancel_timer();
            if (remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
                (void)participant_stateless_message_writer_history_->remove_change(
                    remote_participant_info->change_sequence_number_);
                remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();
            }
        }
        // Restore state machine
        remote_participant_info->auth_status_ = AUTHENTICATION_WAITING_REQUEST;
        // Return the handshake handle
        if (remote_participant_info->handshake_handle_ != nullptr) {
            authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_, exception);
            remote_participant_info->handshake_handle_ = nullptr;
        }
    }

    if (undiscovered && (remote_participant_info->auth_status_ == AUTHENTICATION_REQUEST_NOT_SEND ||
                         remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST)) {
        logInfo(SECURITY, "Resend announceParticipantState to participant " << remote_participant_data.m_guid);
        participant_->pdp()->announceParticipantState(false, false);
    }

    bool returnedValue = true;

    if (remote_participant_info->auth_status_ == AUTHENTICATION_REQUEST_NOT_SEND) {
        // Maybe send request.
        returnedValue = on_process_handshake(remote_participant_data, remote_participant_info, MessageIdentity(),
                                             HandshakeMessageToken(), notify_part_authorized);
    }

    (void)restore_discovered_participant_info(remote_participant_data.m_guid, remote_participant_info);
    if (notify_part_authorized) {
        notify_participant_authorized(remote_participant_data);
    }

    return returnedValue;
}

void SecurityManager::remove_participant(const ParticipantProxyData& participant_data) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    // Unmatch from builtin endpoints.
    unmatch_builtin_endpoints(participant_data);

    // Declare a unique_ptr to hold the authentication info outside the lock scope.
    // NOTE: If tmp_auth is destructed inside the 'sl' (shared_lock) scope, it may cause a deadlock with the timer thread.
    // This is because the destructor of AuthenticationInfo (or related objects) might interact with timer resources
    // that also require acquiring the same mutex, leading to a potential deadlock situation.
    // By moving the destruction of tmp_auth outside the lock, we avoid this risk.
    DiscoveredParticipantInfo::AuthUniquePtr tmp_auth;

    {
        vbs::shared_lock<vbs::shared_mutex> sl(mutex_);

        auto dp_it = discovered_participants_.find(participant_data.m_guid);

        if (dp_it != discovered_participants_.end()) {
            SecurityException exception;

            std::shared_ptr<ParticipantCryptoHandle> participant_crypto_handle =
                dp_it->second->get_participant_crypto();
            if (participant_crypto_handle != nullptr) {
                (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(participant_crypto_handle, exception);
            }

            PermissionsHandle* permissions_handle = dp_it->second->get_permissions_handle();
            if (permissions_handle != nullptr) {
                (void)access_plugin_->return_permissions_handle(permissions_handle, exception);
            }

            std::shared_ptr<SecretHandle> shared_secret_handle = dp_it->second->get_shared_secret();

            if (shared_secret_handle != nullptr) {
                (void)authentication_plugin_->return_sharedsecret_handle(shared_secret_handle, exception);
            }
            tmp_auth = dp_it->second->get_auth();
            remove_discovered_participant_info(tmp_auth);
        }
    }

    std::unique_ptr<DiscoveredParticipantInfo> tmp;

    {
        std::lock_guard<vbs::shared_mutex> _(mutex_);

        for (auto& local_reader : reader_handles_) {
            for (auto wit = local_reader.second.associated_writers.begin();
                 local_reader.second.associated_writers.end() != wit;) {
                if (wit->first.guidPrefix == participant_data.m_guid.guidPrefix) {
                    wit = local_reader.second.associated_writers.erase(wit);
                } else {
                    ++wit;
                }
            }
        }

        for (auto& local_writer : writer_handles_) {
            for (auto rit = local_writer.second.associated_readers.begin();
                 local_writer.second.associated_readers.end() != rit;) {
                if (rit->first.guidPrefix == participant_data.m_guid.guidPrefix) {
                    rit = local_writer.second.associated_readers.erase(rit);
                } else {
                    ++rit;
                }
            }
        }

        //Do not release the second element in the map in the write lock, and let it be released after its function is executed.
        auto it = discovered_participants_.find(participant_data.m_guid);
        if (it != discovered_participants_.end()) {
            tmp = std::move(it->second);
        }
        (void)discovered_participants_.erase(participant_data.m_guid);
    }
}

bool SecurityManager::on_process_handshake(const ParticipantProxyData& participant_data,
                                           DiscoveredParticipantInfo::AuthUniquePtr& remote_participant_info,
                                           MessageIdentity&& message_identity, HandshakeMessageToken&& message_in,
                                           bool& notify_part_authorized) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    HandshakeMessageToken* handshake_message = nullptr;
    SecurityException exception;

    ValidationResult_t ret = VALIDATION_FAILED;

    assert(remote_participant_info->identity_handle_ != nullptr);

    logDebug(SECURITY, "Processing handshake from participant " << participant_data.m_guid);

    if (remote_participant_info->auth_status_ == AUTHENTICATION_REQUEST_NOT_SEND) {
        logInfo(SECURITY,
                "Send handshake request from " << participant_->getGuid() << " to " << participant_data.m_guid);
        ret = authentication_plugin_->begin_handshake_request(
            &remote_participant_info->handshake_handle_, &handshake_message, *local_identity_handle_,
            *remote_participant_info->identity_handle_,
            participant_->pdp()->get_participant_proxy_data_serialized(BIGEND), exception);
    } else if (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST) {
        logInfo(SECURITY, "Received handshake request from " << participant_data.m_guid << " to "
                                                             << participant_->getGuid() << " and send reply");
        assert(!remote_participant_info->handshake_handle_);
        ret = authentication_plugin_->begin_handshake_reply(
            &remote_participant_info->handshake_handle_, &handshake_message, std::move(message_in),
            *remote_participant_info->identity_handle_, *local_identity_handle_,
            participant_->pdp()->get_participant_proxy_data_serialized(BIGEND), exception);
    } else if ((remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REPLY) ||
               (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_FINAL)) {
        logInfo(SECURITY,
                "Received handshake reply/final from " << participant_data.m_guid << " to " << participant_->getGuid());
        assert(remote_participant_info->handshake_handle_);
        ret = authentication_plugin_->process_handshake(&handshake_message, std::move(message_in),
                                                        *remote_participant_info->handshake_handle_, exception);
    } else if (remote_participant_info->auth_status_ == AUTHENTICATION_OK) {
        return true;
    }

    if (ret == VALIDATION_FAILED) {
        remote_participant_info->auth_status_ = AUTHENTICATION_FAILED;
        on_validation_failed(participant_data, exception);
        return false;
    }

    assert(remote_participant_info->handshake_handle_ != nullptr);

    // Remove previous change
    remote_participant_info->event_->cancel_timer();
    if (remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
        (void)participant_stateless_message_writer_history_->remove_change(
            remote_participant_info->change_sequence_number_);
        remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();
    }
    int64_t expected_sequence_number = 0;

    bool handshake_message_send = true;

    if ((ret == VALIDATION_PENDING_HANDSHAKE_MESSAGE) || (ret == VALIDATION_OK_WITH_FINAL_MESSAGE)) {
        handshake_message_send = false;

        assert(handshake_message);

        // Send hanshake message

        // Create message
        ParticipantGenericMessage message =
            generate_authentication_message(std::move(message_identity), participant_data.m_guid, *handshake_message);

        CacheChange_t* change = create_change_for_message(message, participant_stateless_message_writer_history_);

        if (change != nullptr) {
            change->readerGUID = participant_data.m_guid;
            CDRMessage_t aux_msg(0);
            aux_msg.wraps = true;
            aux_msg.buffer = change->serializedPayload.data;
            aux_msg.length = change->serializedPayload.length;
            aux_msg.max_size = change->serializedPayload.max_size;

            // Serialize encapsulation
            (void)CDRMessage::addOctet(&aux_msg, 0);
            aux_msg.msg_endian = DEFAULT_ENDIAN;
            change->serializedPayload.encapsulation = PL_DEFAULT_ENCAPSULATION;
            (void)CDRMessage::addOctet(&aux_msg, DEFAULT_ENCAPSULATION);
            (void)CDRMessage::addUInt16(&aux_msg, 0);

            if (CDRMessage::addParticipantGenericMessage(&aux_msg, message)) {
                change->serializedPayload.length = aux_msg.length;
                logDebug(SECURITY, "ParticipantGenericMessage"
                                       << " (" << participant_->getGuid() << "->" << participant_data.m_guid << ")"
                                       << " MaxMessageSize:" << aux_msg.max_size
                                       << " MessageLength:" << aux_msg.length);

                // Send
                logDebug(SECURITY, "Authentication handshake sent to participant " << participant_data.m_guid);
                participant_stateless_message_writer_history_->set_fragments(change);
                if (participant_stateless_message_writer_history_->add_change(change)) {

                    handshake_message_send = true;
                    expected_sequence_number = message.message_identity().sequence_number();
                    remote_participant_info->change_sequence_number_ = change->sequenceNumber;
                    remote_participant_info->handshake_requests_sent_++;
                } else {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "WriterHistory cannot add the CacheChange_t");
                    // Return the handshake handle
                    authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_,
                                                                    exception);
                    remote_participant_info->handshake_handle_ = nullptr;
                    participant_stateless_message_writer_history_->release_change(change);
                }
            } else {
                // Return the handshake handle
                authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_, exception);
                remote_participant_info->handshake_handle_ = nullptr;
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Cannot serialize ParticipantGenericMessage"
                              << " (" << participant_->getGuid() << "->" << participant_data.m_guid << ")"
                              << " MessageLength:" << aux_msg.max_size << " MessagePosition:" << aux_msg.pos);
                participant_stateless_message_writer_history_->release_change(change);
            }
        } else {
            // Return the handshake handle
            authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_, exception);
            remote_participant_info->handshake_handle_ = nullptr;
            elogError(SECURITY, RetCode_t::RETCODE_NO_DATA, "WriterHistory cannot retrieve a CacheChange_t");
        }
    }

    bool returnedValue = false;
    AuthenticationStatus pre_auth_status = remote_participant_info->auth_status_;

    if (handshake_message_send) {
        switch (ret) {
            case VALIDATION_OK:
            case VALIDATION_OK_WITH_FINAL_MESSAGE:
            case VALIDATION_PENDING_HANDSHAKE_MESSAGE: {
                remote_participant_info->auth_status_ = AUTHENTICATION_OK;
                if (ret == VALIDATION_PENDING_HANDSHAKE_MESSAGE) {
                    if (pre_auth_status == AUTHENTICATION_REQUEST_NOT_SEND) {
                        remote_participant_info->auth_status_ = AUTHENTICATION_WAITING_REPLY;
                    } else if (pre_auth_status == AUTHENTICATION_WAITING_REQUEST) {
                        remote_participant_info->auth_status_ = AUTHENTICATION_WAITING_FINAL;
                    }
                }

                // if authentication was finished, starts encryption.
                if (remote_participant_info->auth_status_ == AUTHENTICATION_OK) {
                    std::shared_ptr<SecretHandle> shared_secret_handle = authentication_plugin_->get_shared_secret(
                        *remote_participant_info->handshake_handle_, exception);
                    if (!participant_authorized(participant_data, remote_participant_info, shared_secret_handle)) {
                        (void)authentication_plugin_->return_sharedsecret_handle(shared_secret_handle, exception);
                    } else {
                        notify_part_authorized = true;
                    }
                }

                if (ret == VALIDATION_PENDING_HANDSHAKE_MESSAGE) {
                    remote_participant_info->expected_sequence_number_ = expected_sequence_number;
                    // Avoid DoS attack by exponentially increasing event interval
                    auto time_ms = remote_participant_info->event_->getIntervalMilliSec();

                    (void)remote_participant_info->event_->update_interval_millisec(
                        time_ms * auth_handshake_props_.handshake_resend_period_gain_);

                    remote_participant_info->event_->restart_timer();
                }

                returnedValue = true;
            } break;
            case VALIDATION_PENDING_RETRY:
            // TODO(Ricardo) Send event.
            default:
                break;
        }
    }

    return returnedValue;
}

bool SecurityManager::create_entities() {
    if (create_participant_stateless_message_entities()) {
        if ((crypto_plugin_ == nullptr) || create_participant_volatile_message_secure_entities()) {
            logDebug(SECURITY, "Initialized security manager for participant " << participant_->getGuid());
            return true;
        }

        delete_participant_stateless_message_entities();
    }

    cancel_init();
    return false;
}

void SecurityManager::delete_entities() {
    delete_participant_volatile_message_secure_entities();
    delete_participant_stateless_message_entities();
}

bool SecurityManager::create_participant_stateless_message_entities() {
    create_participant_stateless_message_pool();
    if (create_participant_stateless_message_writer()) {
        if (create_participant_stateless_message_reader()) {
            return true;
        }
        delete_participant_stateless_message_writer();
    }

    delete_participant_stateless_message_pool();
    return false;
}

void SecurityManager::delete_participant_stateless_message_entities() {
    delete_participant_stateless_message_reader();
    delete_participant_stateless_message_writer();
    delete_participant_stateless_message_pool();
}

void SecurityManager::create_participant_stateless_message_pool() {
    participant_stateless_message_writer_hattr_ = {PREALLOCATED_WITH_REALLOC_MEMORY_MODE,
                                                   PARTICIPANT_STATELESS_MESSAGE_PAYLOAD_DEFAULT_SIZE, 1, 100};
    participant_stateless_message_reader_hattr_ = {PREALLOCATED_WITH_REALLOC_MEMORY_MODE,
                                                   PARTICIPANT_STATELESS_MESSAGE_PAYLOAD_DEFAULT_SIZE, 1, 5000};

    BasicPoolConfig cfg {PREALLOCATED_WITH_REALLOC_MEMORY_MODE, PARTICIPANT_STATELESS_MESSAGE_PAYLOAD_DEFAULT_SIZE};
    participant_stateless_message_pool_ = TopicPayloadPoolRegistry::get("DCPSParticipantStatelessMessage", cfg);

    PoolConfig writer_cfg = PoolConfig::from_history_attributes(participant_stateless_message_writer_hattr_);
    (void)participant_stateless_message_pool_->reserve_history(writer_cfg, false);

    PoolConfig reader_cfg = PoolConfig::from_history_attributes(participant_stateless_message_reader_hattr_);
    (void)participant_stateless_message_pool_->reserve_history(reader_cfg, true);
}

void SecurityManager::delete_participant_stateless_message_pool() {
    if (participant_stateless_message_pool_) {
        PoolConfig writer_cfg = PoolConfig::from_history_attributes(participant_stateless_message_writer_hattr_);
        (void)participant_stateless_message_pool_->release_history(writer_cfg, false);

        PoolConfig reader_cfg = PoolConfig::from_history_attributes(participant_stateless_message_reader_hattr_);
        (void)participant_stateless_message_pool_->release_history(reader_cfg, true);

        participant_stateless_message_pool_.reset();
    }
}

bool SecurityManager::create_participant_stateless_message_writer() {
    participant_stateless_message_writer_history_ = new WriterHistory(participant_stateless_message_writer_hattr_);

    const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();

    WriterAttributes watt;
    watt.endpoint.external_unicast_locators = pattr.builtin.metatraffic_external_unicast_locators;
    watt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    watt.endpoint.endpointKind = WRITER;
    watt.endpoint.reliabilityKind = BEST_EFFORT;
    watt.endpoint.topicKind = NO_KEY;
    watt.matched_readers_allocation = pattr.allocation.participants;

    RTPSWriter* wout = nullptr;
    if (participant_->createWriter(&wout, watt, participant_stateless_message_pool_,
                                   participant_stateless_message_writer_history_, nullptr,
                                   participant_stateless_message_writer_entity_id, true)) {
        if (wout == nullptr) {
            return false;
        }
        participant_->set_endpoint_rtps_protection_supports(wout, false);
        participant_stateless_message_writer_ = dynamic_cast<StatelessWriter*>(wout);
        if (participant_stateless_message_writer_ != nullptr) {
            participant_stateless_message_writer_->set_separate_sending(true);
            auth_source_guid = participant_stateless_message_writer_->getGuid();
        }

        return true;
    }

    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Participant Stateless Message Writer creation failed");
    delete (participant_stateless_message_writer_history_);
    participant_stateless_message_writer_history_ = nullptr;

    return false;
}

void SecurityManager::delete_participant_stateless_message_writer() {
    if (participant_stateless_message_writer_ != nullptr) {
        (void)participant_->deleteUserEndpoint(participant_stateless_message_writer_->getGuid());
        participant_stateless_message_writer_ = nullptr;
    }

    if (participant_stateless_message_writer_history_ != nullptr) {
        delete participant_stateless_message_writer_history_;
        participant_stateless_message_writer_history_ = nullptr;
    }
}

bool SecurityManager::create_participant_stateless_message_reader() {
    participant_stateless_message_reader_history_ = new ReaderHistory(participant_stateless_message_reader_hattr_);

    const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();

    ReaderAttributes ratt;
    ratt.endpoint.topicKind = NO_KEY;
    ratt.endpoint.reliabilityKind = BEST_EFFORT;
    if (!pattr.builtin.avoid_builtin_multicast) {
        ratt.endpoint.multicastLocatorList = pattr.builtin.metatrafficMulticastLocatorList;
    }
    ratt.endpoint.unicastLocatorList = pattr.builtin.metatrafficUnicastLocatorList;
    ratt.endpoint.remoteLocatorList = pattr.builtin.initialPeersList;
    ratt.endpoint.external_unicast_locators = pattr.builtin.metatraffic_external_unicast_locators;
    ratt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    ratt.matched_writers_allocation = pattr.allocation.participants;

    RTPSReader* rout = nullptr;
    if (participant_->createReader(
            &rout, ratt, participant_stateless_message_pool_, participant_stateless_message_reader_history_,
            &participant_stateless_message_listener_, participant_stateless_message_reader_entity_id, true, true)) {
        participant_->set_endpoint_rtps_protection_supports(rout, false);
        participant_stateless_message_reader_ = dynamic_cast<StatelessReader*>(rout);

        return true;
    }

    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Participant Stateless Message Reader creation failed");
    delete (participant_stateless_message_reader_history_);
    participant_stateless_message_reader_history_ = nullptr;
    return false;
}

void SecurityManager::delete_participant_stateless_message_reader() {
    if (participant_stateless_message_reader_ != nullptr) {
        (void)participant_->deleteUserEndpoint(participant_stateless_message_reader_->getGuid());
        participant_stateless_message_reader_ = nullptr;
    }

    if (participant_stateless_message_reader_history_ != nullptr) {
        delete participant_stateless_message_reader_history_;
        participant_stateless_message_reader_history_ = nullptr;
    }
}

bool SecurityManager::create_participant_volatile_message_secure_entities() {
    create_participant_volatile_message_secure_pool();

    if (create_participant_volatile_message_secure_writer()) {
        if (create_participant_volatile_message_secure_reader()) {
            return true;
        }

        delete_participant_volatile_message_secure_writer();
    }

    delete_participant_volatile_message_secure_pool();
    return false;
}

void SecurityManager::delete_participant_volatile_message_secure_entities() {
    delete_participant_volatile_message_secure_reader();
    delete_participant_volatile_message_secure_writer();
    delete_participant_volatile_message_secure_pool();
}

void SecurityManager::create_participant_volatile_message_secure_pool() {
    participant_volatile_message_secure_hattr_ = {PREALLOCATED_WITH_REALLOC_MEMORY_MODE,
                                                  PARTICIPANT_VOLATILE_MESSAGE_PAYLOAD_DEFAULT_SIZE, 1, 0};

    PoolConfig pool_cfg = PoolConfig::from_history_attributes(participant_volatile_message_secure_hattr_);
    participant_volatile_message_secure_pool_ =
        TopicPayloadPoolRegistry::get("DCPSParticipantVolatileMessageSecure", pool_cfg);
    (void)participant_volatile_message_secure_pool_->reserve_history(pool_cfg, false);
    (void)participant_volatile_message_secure_pool_->reserve_history(pool_cfg, true);
}

void SecurityManager::delete_participant_volatile_message_secure_pool() {
    if (participant_volatile_message_secure_pool_) {
        PoolConfig pool_cfg = PoolConfig::from_history_attributes(participant_volatile_message_secure_hattr_);
        (void)participant_volatile_message_secure_pool_->release_history(pool_cfg, true);
        (void)participant_volatile_message_secure_pool_->release_history(pool_cfg, false);
        participant_volatile_message_secure_pool_.reset();
    }
}

bool SecurityManager::create_participant_volatile_message_secure_writer() {
    participant_volatile_message_secure_writer_history_ = new WriterHistory(participant_volatile_message_secure_hattr_);

    const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();

    WriterAttributes watt;
    watt.endpoint.endpointKind = WRITER;
    watt.endpoint.reliabilityKind = RELIABLE;
    watt.endpoint.topicKind = NO_KEY;
    watt.endpoint.durabilityKind = VOLATILE;
    watt.endpoint.unicastLocatorList = pattr.builtin.metatrafficUnicastLocatorList;
    watt.endpoint.external_unicast_locators = pattr.builtin.metatraffic_external_unicast_locators;
    watt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    watt.endpoint.remoteLocatorList = pattr.builtin.initialPeersList;
    watt.endpoint.security_attributes().is_submessage_protected = true;
    watt.endpoint.security_attributes().plugin_endpoint_attributes =
        PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
    watt.matched_readers_allocation = pattr.allocation.participants;

    RTPSWriter* wout = nullptr;
    if (participant_->createWriter(&wout, watt, participant_volatile_message_secure_pool_,
                                   participant_volatile_message_secure_writer_history_, this,
                                   participant_volatile_message_secure_writer_entity_id, true)) {
        if (wout == nullptr) {
            return false;
        }
        participant_->set_endpoint_rtps_protection_supports(wout, false);
        participant_volatile_message_secure_writer_ = dynamic_cast<StatefulWriter*>(wout);
        if (participant_volatile_message_secure_writer_ != nullptr) {
            participant_volatile_message_secure_writer_->set_separate_sending(true);
        }
        return true;
    }

    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Participant Volatile Message Writer creation failed");
    delete (participant_volatile_message_secure_writer_history_);
    participant_volatile_message_secure_writer_history_ = nullptr;

    return false;
}

void SecurityManager::delete_participant_volatile_message_secure_writer() {
    if (participant_volatile_message_secure_writer_ != nullptr) {
        (void)participant_->deleteUserEndpoint(participant_volatile_message_secure_writer_->getGuid());
        participant_volatile_message_secure_writer_ = nullptr;
    }

    if (participant_volatile_message_secure_writer_history_ != nullptr) {
        delete participant_volatile_message_secure_writer_history_;
        participant_volatile_message_secure_writer_history_ = nullptr;
    }
}

bool SecurityManager::create_participant_volatile_message_secure_reader() {
    participant_volatile_message_secure_reader_history_ = new ReaderHistory(participant_volatile_message_secure_hattr_);

    const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();

    ReaderAttributes ratt;
    ratt.endpoint.topicKind = NO_KEY;
    ratt.endpoint.reliabilityKind = RELIABLE;
    ratt.endpoint.durabilityKind = VOLATILE;
    ratt.endpoint.unicastLocatorList = pattr.builtin.metatrafficUnicastLocatorList;
    ratt.endpoint.external_unicast_locators = pattr.builtin.metatraffic_external_unicast_locators;
    ratt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    ratt.endpoint.remoteLocatorList = pattr.builtin.initialPeersList;
    ratt.endpoint.security_attributes().is_submessage_protected = true;
    ratt.endpoint.security_attributes().plugin_endpoint_attributes =
        PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
    ratt.matched_writers_allocation = pattr.allocation.participants;

    RTPSReader* rout = nullptr;
    if (participant_->createReader(&rout, ratt, participant_volatile_message_secure_pool_,
                                   participant_volatile_message_secure_reader_history_,
                                   &participant_volatile_message_secure_listener_,
                                   participant_volatile_message_secure_reader_entity_id, true, true)) {
        participant_->set_endpoint_rtps_protection_supports(rout, false);
        participant_volatile_message_secure_reader_ = dynamic_cast<StatefulReader*>(rout);

        return true;
    }

    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Participant Volatile Message Reader creation failed");
    delete (participant_volatile_message_secure_reader_history_);
    participant_volatile_message_secure_reader_history_ = nullptr;
    return false;
}

void SecurityManager::delete_participant_volatile_message_secure_reader() {
    if (participant_volatile_message_secure_reader_ != nullptr) {
        (void)participant_->deleteUserEndpoint(participant_volatile_message_secure_reader_->getGuid());
        participant_volatile_message_secure_reader_ = nullptr;
    }

    if (participant_volatile_message_secure_reader_history_ != nullptr) {
        delete participant_volatile_message_secure_reader_history_;
        participant_volatile_message_secure_reader_history_ = nullptr;
    }
}

ParticipantGenericMessage SecurityManager::generate_authentication_message(
    const MessageIdentity& related_message_identity, const GUID_t& destination_participant_key,
    HandshakeMessageToken& handshake_message) const {
    ParticipantGenericMessage message;

    message.message_identity().source_guid(auth_source_guid);
    message.message_identity().sequence_number(auth_last_sequence_number_.fetch_add(1));
    message.related_message_identity(related_message_identity);
    message.destination_participant_key(destination_participant_key);
    message.message_class_id(AUTHENTICATION_PARTICIPANT_STATELESS_MESSAGE);
    message.message_data().push_back(handshake_message);

    return message;
}

ParticipantGenericMessage SecurityManager::generate_participant_crypto_token_message(
    const GUID_t& destination_participant_key, ParticipantCryptoTokenSeq& crypto_tokens) const {
    ParticipantGenericMessage message;

    message.message_identity().source_guid(auth_source_guid);
    message.message_identity().sequence_number(crypto_last_sequence_number_.fetch_add(1));
    message.destination_participant_key(destination_participant_key);
    message.message_class_id(GMCLASSID_SECURITY_PARTICIPANT_CRYPTO_TOKENS);
    message.message_data().assign(crypto_tokens.begin(), crypto_tokens.end());

    return message;
}

ParticipantGenericMessage SecurityManager::generate_writer_crypto_token_message(
    const GUID_t& destination_participant_key, const GUID_t& destination_endpoint_key,
    const GUID_t& source_endpoint_key, ParticipantCryptoTokenSeq& crypto_tokens) const {
    ParticipantGenericMessage message;

    message.message_identity().source_guid(auth_source_guid);
    message.message_identity().sequence_number(crypto_last_sequence_number_.fetch_add(1));
    message.destination_participant_key(destination_participant_key);
    message.destination_endpoint_key(destination_endpoint_key);
    message.source_endpoint_key(source_endpoint_key);
    message.message_class_id(GMCLASSID_SECURITY_WRITER_CRYPTO_TOKENS);
    message.message_data().assign(crypto_tokens.begin(), crypto_tokens.end());

    return message;
}

ParticipantGenericMessage SecurityManager::generate_reader_crypto_token_message(
    const GUID_t& destination_participant_key, const GUID_t& destination_endpoint_key,
    const GUID_t& source_endpoint_key, ParticipantCryptoTokenSeq& crypto_tokens) const {
    ParticipantGenericMessage message;

    message.message_identity().source_guid(auth_source_guid);
    message.message_identity().sequence_number(crypto_last_sequence_number_.fetch_add(1));
    message.destination_participant_key(destination_participant_key);
    message.destination_endpoint_key(destination_endpoint_key);
    message.source_endpoint_key(source_endpoint_key);
    message.message_class_id(GMCLASSID_SECURITY_READER_CRYPTO_TOKENS);
    message.message_data().assign(crypto_tokens.begin(), crypto_tokens.end());

    return message;
}

void SecurityManager::process_participant_stateless_message(const CacheChange_t* const change) {
    // std::lock_guard<std::mutex> lck(mtx);
    assert(change);

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    // Deserialize message
    ParticipantGenericMessage message;
    CDRMessage_t aux_msg(0);
    aux_msg.wraps = true;
    aux_msg.buffer = change->serializedPayload.data;
    aux_msg.length = change->serializedPayload.length;
    aux_msg.max_size = change->serializedPayload.max_size;

    // Read encapsulation
    aux_msg.pos += 1;
    octet encapsulation = 0;
    (void)CDRMessage::readOctet(&aux_msg, &encapsulation);
    if (encapsulation == CDR_BE) {
        aux_msg.msg_endian = BIGEND;
    } else if (encapsulation == CDR_LE) {
        aux_msg.msg_endian = LITTLEEND;
    } else {
        return;
    }

    aux_msg.pos += 2;

    (void)CDRMessage::readParticipantGenericMessage(&aux_msg, message);

    if (message.message_class_id().compare(AUTHENTICATION_PARTICIPANT_STATELESS_MESSAGE) == 0) {
        if (message.message_identity().source_guid() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. message_identity.source_guid is GUID_t::unknown()");
            return;
        }
        if (message.destination_participant_key() != participant_->getGuid()) {
            logDebug(SECURITY, "Destination of ParticipantGenericMessage is not me");
            return;
        }
        // Received handshake request log
        if (message.message_data().size() == 1 &&
            message.message_data().at(0).class_id().compare("DDS:Auth:PKI-DH:1.0+Req") == 0) {
            logInfo(SECURITY,
                    "Begin receive handshake request from:" << message.message_identity().source_guid()
                                                            << " to:" << message.destination_participant_key());
        }
        if (message.destination_endpoint_key() != GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. destination_endpoint_key is not GUID_t::unknown()");
            return;
        }
        if (message.source_endpoint_key() != GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. source_endpoint_key is not GUID_t::unknown()");
            return;
        }

        const GUID_t remote_participant_key(message.message_identity().source_guid().guidPrefix,
                                            c_EntityId_RTPSParticipant);
        DiscoveredParticipantInfo::AuthUniquePtr remote_participant_info;
        const ParticipantProxyData* participant_data = nullptr;
        ParticipantProxyData pdata;

        {
            vbs::shared_lock<vbs::shared_mutex> _(mutex_);

            auto dp_it = discovered_participants_.find(remote_participant_key);
            if (dp_it != discovered_participants_.end()) {
                remote_participant_info = dp_it->second->get_auth();
                pdata = dp_it->second->participant_data();
                participant_data = &pdata;
            }
        }

        if (remote_participant_info && participant_data) {
            // Received a handshake request from larger GUID
            if (remote_participant_info->auth_status_ != AUTHENTICATION_OK && message.message_data().size() == 1 &&
                message.message_data().at(0).class_id().compare("DDS:Auth:PKI-DH:1.0+Req") == 0 &&
                message.destination_participant_key() < message.related_message_identity().source_guid()) {
                logInfo(SECURITY, "Received unexpected handshake request from larger GUID:"
                                      << message.related_message_identity().source_guid()
                                      << " to:" << message.destination_participant_key()
                                      << " auth status:" << remote_participant_info->auth_status_);
                (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                return;
            }
            if (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST) {
                assert(!remote_participant_info->handshake_handle_);

                // Preconditions
                if (message.related_message_identity().source_guid() != GUID_t::unknown()) {
                    logDebug(SECURITY,
                             "Bad ParticipantGenericMessage. "
                             "related_message_identity.source_guid is not GUID_t::unknown()");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
                if (message.message_data().size() != 1) {
                    logDebug(SECURITY, "Bad ParticipantGenericMessage. message_data size is not 1");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
            } else if ((remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REPLY) ||
                       (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_FINAL)) {
                assert(remote_participant_info->handshake_handle_);

                if ((message.related_message_identity().source_guid() == GUID_t::unknown()) &&
                    (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_FINAL)) {
                    // Maybe the reply was missed. Resent.
                    if (remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
                        logDebug(SECURITY, "remote_participant_info->change_sequence_number_："
                                               << remote_participant_info->change_sequence_number_);
                        std::lock_guard<std::mutex> lck(mtx);
                        // Remove previous change and send a new one.
                        CacheChange_t* p_change =
                            participant_stateless_message_writer_history_->remove_change_and_reuse(
                                remote_participant_info->change_sequence_number_);
                        remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();

                        if (p_change != nullptr) {
                            participant_stateless_message_writer_history_->set_fragments(p_change);
                            if (participant_stateless_message_writer_history_->add_change(p_change)) {

                                remote_participant_info->change_sequence_number_ = p_change->sequenceNumber;
                                remote_participant_info->handshake_requests_sent_++;
                            } else {
                                logWarning(SECURITY, "add change failed");
                            }
                            // TODO (Ricardo) What to do if not added?
                        }

                        (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                        return;
                    } else {
                        logWarning(SECURITY, "remote_participant_info->change_sequence_number_ is unknown");
                    }
                }

                // Preconditions
                if (message.related_message_identity().source_guid() !=
                    participant_stateless_message_writer_->getGuid()) {
                    logDebug(SECURITY,
                             "Bad ParticipantGenericMessage. "
                             "related_message_identity.source_guid is not mine");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
                if (message.related_message_identity().sequence_number() !=
                    remote_participant_info->expected_sequence_number_) {
                    logDebug(SECURITY,
                             "Bad ParticipantGenericMessage. "
                             "related_message_identity.sequence_number is not expected");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
                if (message.message_data().size() != 1) {
                    logDebug(SECURITY, "Bad ParticipantGenericMessage. message_data size is not");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
            } else if (remote_participant_info->auth_status_ == AUTHENTICATION_OK) {  // LCOV_EXCL_START
                // Clear and re-authentication
                if (message.message_data().size() == 1 &&
                    message.message_data().at(0).class_id().compare("DDS:Auth:PKI-DH:1.0+Req") == 0) {
                    logInfo(SECURITY, "Need reauthentication(AUTH_OK) for:" << message.message_identity().source_guid()
                                                                            << " local:"
                                                                            << message.destination_participant_key());
                    // Remove ParticipantProxyData and its secure info
                    remove_discovered_participant_info(remote_participant_info);
                    // remove_participant(*participant_data);
                    participant_->pdp()->remove_remote_participant(participant_data->m_guid,
                                                                   ParticipantDiscoveryInfo::REMOVED_PARTICIPANT);
                    return;
                }
                // Preconditions
                if (message.related_message_identity().source_guid() !=
                    participant_stateless_message_writer_->getGuid()) {
                    logDebug(SECURITY,
                             "Bad ParticipantGenericMessage. "
                             "related_message_identity.source_guid is not mine");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
                if (message.related_message_identity().sequence_number() !=
                    remote_participant_info->expected_sequence_number_) {
                    logDebug(SECURITY,
                             "Bad ParticipantGenericMessage. "
                             "related_message_identity.sequence_number is not expected");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
                if (message.message_data().size() != 1) {
                    logDebug(SECURITY, "Bad ParticipantGenericMessage. message_data size is not");
                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }

                // Maybe final message was missed. Resent.
                if (remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
                    std::lock_guard<std::mutex> lck(mtx);
                    // Remove previous change and send a new one.
                    CacheChange_t* p_change = participant_stateless_message_writer_history_->remove_change_and_reuse(
                        remote_participant_info->change_sequence_number_);
                    remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();

                    if (p_change != nullptr) {
                        participant_stateless_message_writer_history_->set_fragments(p_change);
                        if (participant_stateless_message_writer_history_->add_change(p_change)) {
                            remote_participant_info->change_sequence_number_ = p_change->sequenceNumber;
                            remote_participant_info->handshake_requests_sent_++;
                        } else {
                            logWarning(SECURITY, "add change failed");
                        }
                        // TODO (Ricardo) What to do if not added?
                    }

                    (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                    return;
                }
            } else {
                (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);
                return;
            }  // LCOV_EXCL_STOP

            bool notify_part_authorized = false;
            (void)on_process_handshake(*participant_data, remote_participant_info,
                                       std::move(message.message_identity()), std::move(message.message_data().at(0)),
                                       notify_part_authorized);

            (void)restore_discovered_participant_info(remote_participant_key, remote_participant_info);

            if (notify_part_authorized) {
                notify_participant_authorized(*participant_data);
            }
        } else {
            logInfo(SECURITY, "Received Authentication message but not found related remote_participant_key:"
                                  << remote_participant_key << " to:" << participant_->getGuid()
                                  << " remote_participant_info ptr:" << (remote_participant_info ? true : false)
                                  << " participant_data ptr:" << participant_data);
        }
    } else {
        logDebug(SECURITY, "Discarted ParticipantGenericMessage with class id " << message.message_class_id());
    }
}

void SecurityManager::process_participant_volatile_message_secure(const CacheChange_t* const change) {
    assert(change);

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    // Deserialize message
    ParticipantGenericMessage message;
    CDRMessage_t aux_msg(0);
    aux_msg.wraps = true;
    aux_msg.buffer = change->serializedPayload.data;
    aux_msg.length = change->serializedPayload.length;
    aux_msg.max_size = change->serializedPayload.max_size;

    // Read encapsulation
    aux_msg.pos += 1;
    octet encapsulation = 0;
    (void)CDRMessage::readOctet(&aux_msg, &encapsulation);
    if (encapsulation == CDR_BE) {
        aux_msg.msg_endian = BIGEND;
    } else if (encapsulation == CDR_LE) {
        aux_msg.msg_endian = LITTLEEND;
    } else {
        return;
    }

    aux_msg.pos += 2;

    (void)CDRMessage::readParticipantGenericMessage(&aux_msg, message);

    if (message.message_class_id().compare(GMCLASSID_SECURITY_PARTICIPANT_CRYPTO_TOKENS) == 0) {
        if (message.message_identity().source_guid() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. message_identity.source_guid is GUID_t::unknown()");
            return;
        }
        if (message.destination_participant_key() != participant_->getGuid()) {
            logDebug(SECURITY, "Destination of ParticipantGenericMessage is not me");
            return;
        }
        if (message.destination_endpoint_key() != GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. destination_endpoint_key is not GUID_t::unknown()");
            return;
        }
        if (message.source_endpoint_key() != GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. source_endpoint_key is not GUID_t::unknown()");
            return;
        }

        const GUID_t remote_participant_key(message.message_identity().source_guid().guidPrefix,
                                            c_EntityId_RTPSParticipant);
        std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto;
        DiscoveredParticipantInfo::AuthUniquePtr remote_participant_info;

        // Search remote participant crypto handle.
        {
            vbs::shared_lock<vbs::shared_mutex> lock(mutex_);

            auto dp_it = discovered_participants_.find(remote_participant_key);

            if (dp_it != discovered_participants_.end()) {
                if (dp_it->second->get_participant_crypto() == nullptr) {
                    return;
                }

                remote_participant_crypto = dp_it->second->get_participant_crypto();
                remote_participant_info = dp_it->second->get_auth();
            } else {
                logDebug(SECURITY,
                         "Received Participant Cryptography message but not found related "
                         "remote_participant_key");
            }
        }

        if (remote_participant_crypto != nullptr) {
            SecurityException exception;

            if (!crypto_plugin_->cryptokeyexchange()->set_remote_participant_crypto_tokens(
                    *local_participant_crypto_handle_, *remote_participant_crypto, message.message_data(), exception)) {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Cannot set remote participant crypto tokens (" << remote_participant_key << ") - ("
                                                                          << exception.what() << ")");
            } else {
                // Release the change from the participant_stateless_message_writer_pool_
                // As both participants have already authorized each other

                if (remote_participant_info &&
                    remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
                    participant_stateless_message_writer_history_->remove_change(
                        remote_participant_info->change_sequence_number_);
                    remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();
                }
            }
        } else {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            (void)remote_participant_pending_messages_.emplace(remote_participant_key,
                                                               std::move(message.message_data()));
        }

        if (remote_participant_info) {
            restore_discovered_participant_info(remote_participant_key, remote_participant_info);
        }
    } else if (message.message_class_id().compare(GMCLASSID_SECURITY_READER_CRYPTO_TOKENS) == 0) {
        if (message.message_identity().source_guid() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. message_identity.source_guid is GUID_t::unknown()");
            return;
        }
        if (message.destination_participant_key() != participant_->getGuid()) {
            logDebug(SECURITY, "Destination of ParticipantGenericMessage is not me");
            return;
        }
        if (message.destination_endpoint_key() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. destination_endpoint_key is GUID_t::unknown()");
            return;
        }
        if (message.source_endpoint_key() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. source_endpoint_key is GUID_t::unknown()");
            return;
        }

        // Search remote writer handle.
        GUID_t writer_guid;
        ReaderProxyData* reader_data = nullptr;

        {
            std::lock_guard<vbs::shared_mutex> _(mutex_);

            auto wr_it = writer_handles_.find(message.destination_endpoint_key());

            if (wr_it != writer_handles_.end()) {
                auto rd_it = wr_it->second.associated_readers.find(message.source_endpoint_key());

                if (rd_it != wr_it->second.associated_readers.end()) {
                    SecurityException exception;

                    if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                            *wr_it->second.writer_handle, *std::get<1>(rd_it->second), message.message_data(),
                            exception)) {
                        writer_guid = wr_it->first;
                        reader_data = &(std::get<0>(rd_it->second));
                        logDebug(SECURITY,
                                 "ReaderProxyData for pair:" << writer_guid << "-" << message.source_endpoint_key()
                                                             << " unicastlocatorssize:"
                                                             << reader_data->remote_locators().unicast.size());
                    } else {
                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                  "Cannot set remote reader crypto tokens (" << message.source_endpoint_key() << ") - ("
                                                                             << exception.what() << ")");
                    }
                } else {
                    (void)remote_reader_pending_messages_.emplace(
                        std::make_pair(message.source_endpoint_key(), message.destination_endpoint_key()),
                        std::move(message.message_data()));
                }
            } else {
                logDebug(SECURITY, "Received Reader Cryptography message but not found local writer "
                                       << message.destination_endpoint_key());
            }
        }

        // If writer was found and setting of crypto tokens works, then tell core to match writer
        // and reader.
        if (writer_guid != GUID_t::unknown()) {
            (void)participant_->pairing_remote_reader_with_local_writer_after_security(writer_guid, *reader_data);
        }
    } else if (message.message_class_id().compare(GMCLASSID_SECURITY_WRITER_CRYPTO_TOKENS) == 0) {
        if (message.message_identity().source_guid() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. message_identity.source_guid is GUID_t::unknown()");
            return;
        }
        if (message.destination_participant_key() != participant_->getGuid()) {
            logDebug(SECURITY, "Destination of ParticipantGenericMessage is not me");
            return;
        }
        if (message.destination_endpoint_key() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. destination_endpoint_key is GUID_t::unknown()");
            return;
        }
        if (message.source_endpoint_key() == GUID_t::unknown()) {
            logDebug(SECURITY, "Bad ParticipantGenericMessage. source_endpoint_key is GUID_t::unknown()");
            return;
        }

        // Search remote writer handle.
        GUID_t reader_guid;
        WriterProxyData* writer_data = nullptr;

        {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            auto rd_it = reader_handles_.find(message.destination_endpoint_key());

            if (rd_it != reader_handles_.end()) {
                auto wr_it = rd_it->second.associated_writers.find(message.source_endpoint_key());

                if (wr_it != rd_it->second.associated_writers.end()) {
                    SecurityException exception;

                    if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                            *rd_it->second.reader_handle, *std::get<1>(wr_it->second), message.message_data(),
                            exception)) {
                        reader_guid = rd_it->first;
                        writer_data = &(std::get<0>(wr_it->second));
                    } else {
                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                  "Cannot set remote writer crypto tokens (" << message.source_endpoint_key() << ") - ("
                                                                             << exception.what() << ")");
                    }
                } else {
                    (void)remote_writer_pending_messages_.emplace(
                        std::make_pair(message.source_endpoint_key(), message.destination_endpoint_key()),
                        std::move(message.message_data()));
                }
            } else {
                logDebug(SECURITY, "Received Writer Cryptography message but not found local reader "
                                       << message.destination_endpoint_key());
            }
        }

        // If reader was found and setting of crypto tokens works, then tell core to match reader
        // and writer.
        if (reader_guid != GUID_t::unknown()) {
            (void)participant_->pairing_remote_writer_with_local_reader_after_security(reader_guid, *writer_data);
        }
    } else {
        logDebug(SECURITY, "Discarted ParticipantGenericMessage with class id " << message.message_class_id());
    }
}
void SecurityManager::ParticipantStatelessMessageListener::onNewCacheChangeAdded(
    vbs::dispatcher::SubstrateReader* sub_reader, const CacheChange_t* const change) {
    if (change->readerGUID.entityId == participant_stateless_message_reader_entity_id) {
        std::ostringstream oss;
        oss << change->writerGUID << " " << change->readerGUID;
        elogInfoKeyT(SECURITY, oss.str(),
                     "New change added handshake message from " << change->writerGUID << " to " << change->readerGUID
                                                                << " length " << change->serializedPayload.length);
    }
    manager_.process_participant_stateless_message(change);

    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);
    ReaderHistory* history = reader->getHistory();
    assert(history);
    (void)history->remove_change(const_cast<CacheChange_t*>(change));
}

void SecurityManager::ParticipantVolatileMessageListener::onNewCacheChangeAdded(
    vbs::dispatcher::SubstrateReader* sub_reader, const CacheChange_t* const change) {
    manager_.process_participant_volatile_message_secure(change);

    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);
    ReaderHistory* history = reader->getHistory();
    assert(history);
    (void)history->remove_change(const_cast<CacheChange_t*>(change));
}

bool SecurityManager::get_identity_token(IdentityToken** identity_token) const {
    assert(identity_token);

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (authentication_plugin_) {
        SecurityException exception;
        return authentication_plugin_->get_identity_token(identity_token, *local_identity_handle_, exception);
    }

    return false;
}

bool SecurityManager::return_identity_token(IdentityToken* identity_token) const {
    if (identity_token == nullptr) {
        return true;
    }

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (authentication_plugin_) {
        SecurityException exception;
        return authentication_plugin_->return_identity_token(identity_token, exception);
    }

    return false;
}

bool SecurityManager::get_permissions_token(PermissionsToken** permissions_token) const {
    assert(permissions_token);

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (access_plugin_) {
        SecurityException exception;
        return access_plugin_->get_permissions_token(permissions_token, *local_permissions_handle_, exception);
    }

    return false;
}

bool SecurityManager::return_permissions_token(PermissionsToken* permissions_token) const {
    if (permissions_token == nullptr) {
        return true;
    }

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (access_plugin_) {
        SecurityException exception;
        return access_plugin_->return_permissions_token(permissions_token, exception);
    }

    return false;
}

uint32_t SecurityManager::builtin_endpoints() const {
    uint32_t be = 0;

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return be;
    }

    if (participant_stateless_message_reader_ != nullptr) {
        be |= BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_READER;
    }
    if (participant_stateless_message_writer_ != nullptr) {
        be |= BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_WRITER;
    }
    if (participant_volatile_message_secure_reader_ != nullptr) {
        be |= BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER;
    }
    if (participant_volatile_message_secure_writer_ != nullptr) {
        be |= BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER;
    }

    return be;
}

bool SecurityManager::check_guid_comes_from(const GUID_t& adjusted, const GUID_t& original) const {
    bool ret = (original == adjusted);
    if (!ret && authentication_plugin_ != nullptr) {
        vbs::shared_lock<vbs::shared_mutex> _(mutex_);
        auto part_it = discovered_participants_.find(adjusted);
        if (part_it != discovered_participants_.end()) {
            ret = part_it->second->check_guid_comes_from(authentication_plugin_, adjusted, original);
        }
    }
    return ret;
}

void SecurityManager::match_builtin_endpoints(const ParticipantProxyData& participant_data) {
    uint32_t builtin_endpoints = participant_data.m_availableBuiltinEndpoints;
    bool writer_added = false;
    bool reader_added = false;

    if ((participant_stateless_message_reader_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_WRITER)) {

        auto temp_stateless_writer_proxy_data_ = get_temporary_writer_proxies_pool().get();

        temp_stateless_writer_proxy_data_->clear();
        temp_stateless_writer_proxy_data_->guid().guidPrefix = participant_data.m_guid.guidPrefix;
        temp_stateless_writer_proxy_data_->guid().entityId = participant_stateless_message_writer_entity_id;
        temp_stateless_writer_proxy_data_->persistence_guid(temp_stateless_writer_proxy_data_->guid());
        temp_stateless_writer_proxy_data_->set_remote_locators(participant_data.metatraffic_locators, participant_,
                                                               false);
        temp_stateless_writer_proxy_data_->topicKind(NO_KEY);
        temp_stateless_writer_proxy_data_->m_qos.m_reliability.kind = BEST_EFFORT_RELIABILITY_QOS;
        temp_stateless_writer_proxy_data_->m_qos.m_durability.kind = VOLATILE_DURABILITY_QOS;
        participant_stateless_message_reader_->matched_writer_add(*temp_stateless_writer_proxy_data_);
        writer_added = true;
    }

    if ((participant_stateless_message_writer_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_READER)) {

        auto temp_stateless_reader_proxy_data_ = get_temporary_reader_proxies_pool().get();

        temp_stateless_reader_proxy_data_->clear();
        // temp_stateless_reader_proxy_data_->m_expectsInlineQos = false;
        temp_stateless_reader_proxy_data_->guid().guidPrefix = participant_data.m_guid.guidPrefix;
        temp_stateless_reader_proxy_data_->guid().entityId = participant_stateless_message_reader_entity_id;
        temp_stateless_reader_proxy_data_->set_remote_locators(participant_data.metatraffic_locators, participant_,
                                                               false);
        temp_stateless_reader_proxy_data_->topicKind(NO_KEY);
        temp_stateless_reader_proxy_data_->m_qos.m_reliability.kind = BEST_EFFORT_RELIABILITY_QOS;
        temp_stateless_reader_proxy_data_->m_qos.m_durability.kind = VOLATILE_DURABILITY_QOS;
        participant_stateless_message_writer_->matched_reader_add(*temp_stateless_reader_proxy_data_);
        reader_added = true;
    }

    if (!writer_added || !reader_added) {
        elogError(SECURITY, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Available builtin endpoints mask 0x" << std::hex << builtin_endpoints << std::dec
                                                        << " for remote:" << participant_data.m_guid
                                                        << " local:" << participant_->getGuid());
    }
}

void SecurityManager::match_builtin_key_exchange_endpoints(const ParticipantProxyData& participant_data) {
    uint32_t builtin_endpoints = participant_data.m_availableBuiltinEndpoints;

    if ((participant_volatile_message_secure_reader_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER)) {

        auto temp_volatile_writer_proxy_data_ = get_temporary_writer_proxies_pool().get();

        temp_volatile_writer_proxy_data_->clear();
        temp_volatile_writer_proxy_data_->guid().guidPrefix = participant_data.m_guid.guidPrefix;
        temp_volatile_writer_proxy_data_->guid().entityId = participant_volatile_message_secure_writer_entity_id;
        temp_volatile_writer_proxy_data_->persistence_guid(temp_volatile_writer_proxy_data_->guid());
        temp_volatile_writer_proxy_data_->set_remote_locators(participant_data.metatraffic_locators, participant_,
                                                              false);
        temp_volatile_writer_proxy_data_->topicKind(NO_KEY);
        temp_volatile_writer_proxy_data_->m_qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
        temp_volatile_writer_proxy_data_->m_qos.m_durability.kind = VOLATILE_DURABILITY_QOS;
        participant_volatile_message_secure_reader_->matched_writer_add(*temp_volatile_writer_proxy_data_);
    }

    if ((participant_volatile_message_secure_writer_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER)) {

        auto temp_volatile_reader_proxy_data_ = get_temporary_reader_proxies_pool().get();

        temp_volatile_reader_proxy_data_->clear();
        // temp_volatile_reader_proxy_data_->m_expectsInlineQos = false;
        temp_volatile_reader_proxy_data_->guid().guidPrefix = participant_data.m_guid.guidPrefix;
        temp_volatile_reader_proxy_data_->guid().entityId = participant_volatile_message_secure_reader_entity_id;
        temp_volatile_reader_proxy_data_->set_remote_locators(participant_data.metatraffic_locators, participant_,
                                                              false);
        temp_volatile_reader_proxy_data_->topicKind(NO_KEY);
        temp_volatile_reader_proxy_data_->m_qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
        temp_volatile_reader_proxy_data_->m_qos.m_durability.kind = VOLATILE_DURABILITY_QOS;
        participant_volatile_message_secure_writer_->matched_reader_add(*temp_volatile_reader_proxy_data_);
    }
}

void SecurityManager::unmatch_builtin_endpoints(const ParticipantProxyData& participant_data) {
    uint32_t builtin_endpoints = participant_data.m_availableBuiltinEndpoints;
    GUID_t tmp_guid;
    tmp_guid.guidPrefix = participant_data.m_guid.guidPrefix;

    if ((participant_stateless_message_reader_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_WRITER)) {
        tmp_guid.entityId = participant_stateless_message_writer_entity_id;
        (void)participant_stateless_message_reader_->matched_writer_remove(tmp_guid, false);
    }

    if ((participant_stateless_message_writer_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_STATELESS_MESSAGE_READER)) {
        tmp_guid.entityId = participant_stateless_message_reader_entity_id;
        (void)participant_stateless_message_writer_->matched_reader_remove(tmp_guid);
    }

    if ((participant_volatile_message_secure_reader_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_WRITER)) {
        tmp_guid.entityId = participant_volatile_message_secure_writer_entity_id;
        (void)participant_volatile_message_secure_reader_->matched_writer_remove(tmp_guid, false);
    }

    if ((participant_volatile_message_secure_writer_ != nullptr) &&
        (builtin_endpoints & BUILTIN_ENDPOINT_PARTICIPANT_VOLATILE_MESSAGE_SECURE_READER)) {
        tmp_guid.entityId = participant_volatile_message_secure_reader_entity_id;
        (void)participant_volatile_message_secure_writer_->matched_reader_remove(tmp_guid);
    }
}

void SecurityManager::exchange_participant_crypto(std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto,
                                                  const GUID_t& remote_participant_guid) {
    SecurityException exception;

    // Get participant crypto tokens.
    ParticipantCryptoTokenSeq local_participant_crypto_tokens;
    if (crypto_plugin_->cryptokeyexchange()->create_local_participant_crypto_tokens(
            local_participant_crypto_tokens, *local_participant_crypto_handle_, *remote_participant_crypto,
            exception)) {
        ParticipantGenericMessage message =
            generate_participant_crypto_token_message(remote_participant_guid, local_participant_crypto_tokens);

        CacheChange_t* change = create_change_for_message(message, participant_volatile_message_secure_writer_history_);

        if (change != nullptr) {
            // Serialize message
            CDRMessage_t aux_msg(0);
            aux_msg.wraps = true;
            aux_msg.buffer = change->serializedPayload.data;
            aux_msg.length = change->serializedPayload.length;
            aux_msg.max_size = change->serializedPayload.max_size;

            // Serialize encapsulation
            (void)CDRMessage::addOctet(&aux_msg, 0);
            aux_msg.msg_endian = DEFAULT_ENDIAN;
            change->serializedPayload.encapsulation = PL_DEFAULT_ENCAPSULATION;
            (void)CDRMessage::addOctet(&aux_msg, DEFAULT_ENCAPSULATION);
            (void)CDRMessage::addUInt16(&aux_msg, 0);

            if (CDRMessage::addParticipantGenericMessage(&aux_msg, message)) {
                change->serializedPayload.length = aux_msg.length;

                // Send
                if (!participant_volatile_message_secure_writer_history_->add_change(change)) {
                    (void)participant_volatile_message_secure_writer_history_->release_change(change);
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR, "WriterHistory cannot add the CacheChange_t");
                }
            } else {
                (void)participant_volatile_message_secure_writer_history_->release_change(change);
                elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot serialize ParticipantGenericMessage");
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_NO_DATA, "WriterHistory cannot retrieve a CacheChange_t");
        }
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Error generating crypto token. (" << exception.what() << ")");
    }
}

// TODO (Ricardo) Change participant_data
std::shared_ptr<ParticipantCryptoHandle> SecurityManager::register_and_match_crypto_endpoint(
    IdentityHandle& remote_participant_identity, SecretHandle& shared_secret) {
    if (crypto_plugin_ == nullptr) {
        return nullptr;
    }

    NilHandle nil_handle;
    SecurityException exception;

    // Register remote participant into crypto plugin.
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto =
        crypto_plugin_->cryptokeyfactory()->register_matched_remote_participant(
            *local_participant_crypto_handle_, remote_participant_identity, nil_handle, shared_secret, exception);

    if (remote_participant_crypto != nullptr) {
        return remote_participant_crypto;
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                  "Error registering remote participant in cryptography plugin. (" << exception.what() << ")");
    }

    return nullptr;
}

bool SecurityManager::encode_rtps_message(const CDRMessage_t& input_message, CDRMessage_t& output_message,
                                          const std::vector<GuidPrefix_t>& receiving_list) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    if (crypto_plugin_ == nullptr) {
        try {
            elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED,
                      "Trying to encode rtps message without set cryptography plugin.");
        } catch (...) {}
        return false;
    }

    assert(receiving_list.size() > 0);

    std::vector<std::shared_ptr<ParticipantCryptoHandle>> receiving_crypto_list;
    for (const auto& remote_participant : receiving_list) {
        const GUID_t remote_participant_key(remote_participant, c_EntityId_RTPSParticipant);

        if (remote_participant_key == participant_->getGuid()) {
            receiving_crypto_list.push_back(local_participant_crypto_handle_);
        } else {
            auto dp_it = discovered_participants_.find(remote_participant_key);

            if (dp_it != discovered_participants_.cend() && dp_it->second->get_participant_crypto() != nullptr) {
                receiving_crypto_list.push_back(dp_it->second->get_participant_crypto());
            } else {
                logDebug(SECURITY, "Cannot encode message for participant " << remote_participant_key);
            }
        }
    }

    SecurityException exception;
    return crypto_plugin_->cryptotransform()->encode_rtps_message(
        output_message, input_message, *local_participant_crypto_handle_, receiving_crypto_list, exception);
}

int SecurityManager::decode_rtps_message(const CDRMessage_t& message, CDRMessage_t& out_message,
                                         const GuidPrefix_t& remote_participant) const {
    if (message.buffer[message.pos] != SRTPS_PREFIX) {
        return 1;
    }

    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return 0;
    }

    if (crypto_plugin_ == nullptr) {
        return 0;
    }

    // Init output buffer
    (void)CDRMessage::initCDRMsg(&out_message);

    std::shared_ptr<const ParticipantCryptoHandle> remote_participant_crypto_handle;

    const GUID_t remote_participant_key(remote_participant, c_EntityId_RTPSParticipant);

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        vbs::shared_lock<vbs::shared_mutex> _(mutex_);

        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
        }
    }

    int returnedValue = -1;

    if (remote_participant_crypto_handle != nullptr) {
        SecurityException exception;
        bool ret = crypto_plugin_->cryptotransform()->decode_rtps_message(
            out_message, message, *local_participant_crypto_handle_, *remote_participant_crypto_handle, exception);

        if (ret) {
            returnedValue = 0;
        } else {
            logDebug(SECURITY, "Cannot decode rtps message from participant " << remote_participant_key << "("
                                                                              << exception.what() << ")");
        }
    } else {
        logDebug(SECURITY, "Cannot decode message. Participant " << remote_participant_key << " is not yet authorized");
    }

    return returnedValue;
}

bool SecurityManager::register_local_writer(const GUID_t& writer_guid, const PropertyPolicy& writer_properties,
                                            EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    SecurityException exception;
    bool returned_value = get_datawriter_sec_attributes(writer_properties, security_attributes);

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        DatawriterCryptoHandle* writer_handle = crypto_plugin_->cryptokeyfactory()->register_local_datawriter(
            *local_participant_crypto_handle_, writer_properties.properties(), security_attributes, exception);

        if ((writer_handle != nullptr) && (!writer_handle->nil())) {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            (void)writer_handles_.emplace(writer_guid, writer_handle);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                      "Cannot register local writer in crypto plugin. (" << exception.what() << ")");
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::get_datawriter_sec_attributes(const PropertyPolicy& writer_properties,
                                                    EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    bool returned_value = true;
    SecurityException exception;

    if (access_plugin_ != nullptr) {
        // Extract topic and partitions.
        std::string topic_name, partitions_str;
        std::vector<std::string> partitions;
        const std::string* property_value = PropertyPolicyHelper::find_property(writer_properties, "topic_name");

        if (property_value != nullptr) {
            topic_name = *property_value;
        }

        property_value = PropertyPolicyHelper::find_property(writer_properties, "partitions");

        if (property_value != nullptr) {
            partitions_str = *property_value;

            // Extract partitions.
            std::size_t initial_pos = 0, last_pos = partitions_str.find_first_of(';');
            while (last_pos != std::string::npos) {
                if (initial_pos <= partitions_str.size() && last_pos <= partitions_str.size()) {
                    partitions.emplace_back(partitions_str.begin() + initial_pos, partitions_str.begin() + last_pos);
                }
                initial_pos = last_pos + 1;
                last_pos = partitions_str.find_first_of(';', last_pos + 1);
            }
            if (initial_pos < partitions_str.size()) {
                partitions.emplace_back(partitions_str.begin() + initial_pos, partitions_str.end());
            }
        }

        if (!topic_name.empty()) {
            if (access_plugin_->check_create_datawriter(*local_permissions_handle_, domain_id_, topic_name, partitions,
                                                        exception)) {
                if ((returned_value = access_plugin_->get_datawriter_sec_attributes(
                         *local_permissions_handle_, topic_name, partitions, security_attributes, exception)) ==
                    false) {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Error getting security attributes of local writer "
                                  << " (" << exception.what() << ")" << std::endl);
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking creation of local writer "
                              << " (" << exception.what() << ")" << std::endl);
                returned_value = false;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED, "Error. No topic_name." << std::endl);
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::register_local_builtin_writer(const GUID_t& writer_guid,
                                                    EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    bool returned_value = true;
    SecurityException exception;

    if ((crypto_plugin_ != nullptr) && security_attributes.is_submessage_protected &&
        (writer_guid.entityId != participant_volatile_message_secure_writer_entity_id)) {
        PropertySeq auxProps;
        DatawriterCryptoHandle* writer_handle = crypto_plugin_->cryptokeyfactory()->register_local_datawriter(
            *local_participant_crypto_handle_, auxProps, security_attributes, exception);

        if ((writer_handle != nullptr) && (!writer_handle->nil())) {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            (void)writer_handles_.emplace(writer_guid, writer_handle);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                      "Cannot register local writer in crypto plugin. (" << exception.what() << ")");
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::unregister_local_writer(const GUID_t& writer_guid) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    std::lock_guard<vbs::shared_mutex> _(mutex_);

    auto local_writer = writer_handles_.find(writer_guid);

    if (local_writer != writer_handles_.end()) {
        SecurityException exception;

        for (auto& rit : local_writer->second.associated_readers) {
            (void)crypto_plugin_->cryptokeyfactory()->unregister_datareader(std::get<1>(rit.second), exception);
        }

        (void)crypto_plugin_->cryptokeyfactory()->unregister_datawriter(local_writer->second.writer_handle, exception);
        (void)writer_handles_.erase(local_writer);

        return true;
    }

    return false;
}

bool SecurityManager::register_local_reader(const GUID_t& reader_guid, const PropertyPolicy& reader_properties,
                                            EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    SecurityException exception;
    bool returned_value = get_datareader_sec_attributes(reader_properties, security_attributes);

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        DatareaderCryptoHandle* reader_handle = crypto_plugin_->cryptokeyfactory()->register_local_datareader(
            *local_participant_crypto_handle_, reader_properties.properties(), security_attributes, exception);

        if (reader_handle != nullptr && !reader_handle->nil()) {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            (void)reader_handles_.emplace(reader_guid, reader_handle);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                      "Cannot register local reader in crypto plugin. (" << exception.what() << ")");
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::get_datareader_sec_attributes(const PropertyPolicy& reader_properties,
                                                    EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    bool returned_value = true;
    SecurityException exception;

    if (access_plugin_ != nullptr) {
        // Extract topic and partitions.
        std::string topic_name, partitions_str;
        std::vector<std::string> partitions;
        const std::string* property_value = PropertyPolicyHelper::find_property(reader_properties, "topic_name");

        if (property_value != nullptr) {
            topic_name = *property_value;
        }

        property_value = PropertyPolicyHelper::find_property(reader_properties, "partitions");

        if (property_value != nullptr) {
            partitions_str = *property_value;

            // Extract partitions.
            std::size_t initial_pos = 0, last_pos = partitions_str.find_first_of(';');
            while (last_pos != std::string::npos) {
                partitions.emplace_back(partitions_str.begin() + initial_pos, partitions_str.begin() + last_pos);
                initial_pos = last_pos + 1;
                last_pos = partitions_str.find_first_of(';', last_pos + 1);
            }
            if (initial_pos < partitions_str.size()) {
                partitions.emplace_back(partitions_str.begin() + initial_pos, partitions_str.end());
            }
        }

        if (!topic_name.empty()) {
            if (access_plugin_->check_create_datareader(*local_permissions_handle_, domain_id_, topic_name, partitions,
                                                        exception)) {
                if ((returned_value = access_plugin_->get_datareader_sec_attributes(
                         *local_permissions_handle_, topic_name, partitions, security_attributes, exception)) ==
                    false) {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Error getting security attributes of local reader "
                                  << " (" << exception.what() << ")" << std::endl);
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking creation of local reader "
                              << " (" << exception.what() << ")" << std::endl);
                returned_value = false;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_NOT_ENABLED, "Error. No topic_name." << std::endl);
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::register_local_builtin_reader(const GUID_t& reader_guid,
                                                    EndpointSecurityAttributes& security_attributes) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    bool returned_value = true;
    SecurityException exception;

    if ((crypto_plugin_ != nullptr) && security_attributes.is_submessage_protected &&
        (reader_guid.entityId != participant_volatile_message_secure_reader_entity_id)) {
        PropertySeq auxProps;
        DatareaderCryptoHandle* reader_handle = crypto_plugin_->cryptokeyfactory()->register_local_datareader(
            *local_participant_crypto_handle_, auxProps, security_attributes, exception);

        if ((reader_handle != nullptr) && (!reader_handle->nil())) {
            std::lock_guard<vbs::shared_mutex> _(mutex_);
            (void)reader_handles_.emplace(reader_guid, reader_handle);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                      "Cannot register local reader in crypto plugin. (" << exception.what() << ")");
            returned_value = false;
        }
    }

    return returned_value;
}

bool SecurityManager::unregister_local_reader(const GUID_t& reader_guid) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    std::lock_guard<vbs::shared_mutex> _(mutex_);

    auto local_reader = reader_handles_.find(reader_guid);

    if (local_reader != reader_handles_.end()) {
        SecurityException exception;

        for (auto& wit : local_reader->second.associated_writers) {
            (void)crypto_plugin_->cryptokeyfactory()->unregister_datawriter(std::get<1>(wit.second), exception);
        }

        (void)crypto_plugin_->cryptokeyfactory()->unregister_datareader(local_reader->second.reader_handle, exception);
        (void)reader_handles_.erase(local_reader);

        return true;
    }

    return false;
}

bool SecurityManager::discovered_reader(const GUID_t& writer_guid, const GUID_t& remote_participant_key,
                                        ReaderProxyData& remote_reader_data,
                                        const EndpointSecurityAttributes& security_attributes) {
    return discovered_reader(writer_guid, remote_participant_key, remote_reader_data, security_attributes, false);
}

void SecurityManager::remove_reader(const GUID_t& writer_guid, const GUID_t& /*remote_participant_key*/,
                                    const GUID_t& remote_reader_guid) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    if (crypto_plugin_ == nullptr) {
        return;
    }

    std::lock_guard<vbs::shared_mutex> _(mutex_);

    auto local_writer = writer_handles_.find(writer_guid);

    if (local_writer != writer_handles_.end()) {
        SecurityException exception;

        auto rit = local_writer->second.associated_readers.find(remote_reader_guid);

        if (rit != local_writer->second.associated_readers.end()) {
            (void)crypto_plugin_->cryptokeyfactory()->unregister_datareader(std::get<1>(rit->second), exception);
            (void)local_writer->second.associated_readers.erase(rit);
        } else {
            logDebug(SECURITY, "Cannot find remote reader " << remote_reader_guid << std::endl);
        }
    }
}

bool SecurityManager::discovered_builtin_reader(const GUID_t& writer_guid, const GUID_t& remote_participant_key,
                                                ReaderProxyData& remote_reader_data,
                                                const EndpointSecurityAttributes& security_attributes) {
    return discovered_reader(writer_guid, remote_participant_key, remote_reader_data, security_attributes, true);
}

bool SecurityManager::discovered_reader(const GUID_t& writer_guid, const GUID_t& remote_participant_key,
                                        ReaderProxyData& remote_reader_data,
                                        const EndpointSecurityAttributes& security_attributes, bool is_builtin) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    std::unique_lock<vbs::shared_mutex> lock(mutex_);

    PermissionsHandle* remote_permissions = nullptr;
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto_handle;
    std::shared_ptr<SecretHandle> shared_secret_handle;
    AuthenticationStatus auth_status(AUTHENTICATION_INIT);

    if (!security_attributes.match(remote_reader_data.security_attributes_,
                                   remote_reader_data.plugin_security_attributes_)) {
        logError(SECURITY,
                 "reader security_attributes unmatch with plugin: " << remote_reader_data.security_attributes_ << " "
                                                                    << remote_reader_data.plugin_security_attributes_);
        return false;
    }

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_permissions = dp_it->second->get_permissions_handle();
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
            shared_secret_handle = dp_it->second->get_shared_secret();
            auth_status = dp_it->second->get_auth_status();
        }
    }

    // assert(access_plugin_ == nullptr || remote_permissions != nullptr);
    // assert(crypto_plugin_ == nullptr || remote_participant_crypto_handle != nullptr);

    bool relay_only = false;
    bool returned_value = true;
    SecurityException exception;

    if (!is_builtin) {
        //! Check if it is an unathenticated participant
        if (participant_->security_attributes().allow_unauthenticated_participants &&
            auth_status != AUTHENTICATION_NOT_AVAILABLE && auth_status != AUTHENTICATION_OK &&
            (security_attributes.is_write_protected || security_attributes.is_read_protected)) {
            //!Do not match if read or write protection is enabled for this local endpoint
            logError(SECURITY, "reader security_attributes unmatch with plugin: "
                                   << auth_status << " is_write_protected " << security_attributes.is_write_protected
                                   << " is_read_protected " << security_attributes.is_read_protected);
            return false;
        }

        if (access_plugin_ != nullptr && remote_permissions != nullptr) {
            if ((returned_value = access_plugin_->check_remote_datareader(
                     *remote_permissions, domain_id_, remote_reader_data, relay_only, exception)) == false) {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking create remote reader " << remote_reader_data.guid() << " ("
                                                                 << exception.what() << ")");
            }
        }
    }

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        bool is_key_exchange =
            (remote_reader_data.guid().entityId == participant_volatile_message_secure_reader_entity_id);
        auto local_writer = writer_handles_.find(writer_guid);
        returned_value = false;

        if (local_writer != writer_handles_.end()) {
            if (remote_participant_crypto_handle != nullptr && shared_secret_handle != nullptr) {
                DatareaderCryptoHandle* remote_reader_handle =
                    crypto_plugin_->cryptokeyfactory()->register_matched_remote_datareader(
                        *local_writer->second.writer_handle, *remote_participant_crypto_handle, *shared_secret_handle,
                        relay_only, exception);

                if ((remote_reader_handle != nullptr) && (!remote_reader_handle->nil())) {
                    if (is_key_exchange) {
                        logDebug(SECURITY, "Process successful discovering local reader "
                                               << remote_reader_data.guid() << " unicastlocatorsize:"
                                               << remote_reader_data.remote_locators().unicast.size());
                        (void)local_writer->second.associated_readers.emplace(
                            remote_reader_data.guid(), std::make_tuple(remote_reader_data, remote_reader_handle));
                        lock.unlock();
                        (void)participant_->pairing_remote_reader_with_local_writer_after_security(writer_guid,
                                                                                                   remote_reader_data);
                    } else {
                        // Check pending reader crypto messages.
                        auto pending = remote_reader_pending_messages_.find(
                            std::make_pair(remote_reader_data.guid(), writer_guid));
                        bool pairing_cause_pending_message = false;

                        if (pending != remote_reader_pending_messages_.end()) {
                            if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                                    *local_writer->second.writer_handle, *remote_reader_handle, pending->second,
                                    exception)) {
                                pairing_cause_pending_message = true;
                            } else {
                                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                          "Cannot set remote reader crypto tokens ("
                                              << remote_reader_data.guid() << ") - (" << exception.what() << ")");
                            }

                            (void)remote_reader_pending_messages_.erase(pending);
                        }

                        GUID_t local_reader_guid;
                        WriterProxyData* writer_data = nullptr;

                        // Get local writer crypto tokens.
                        DatawriterCryptoTokenSeq local_writer_crypto_tokens;
                        if (crypto_plugin_->cryptokeyexchange()->create_local_datawriter_crypto_tokens(
                                local_writer_crypto_tokens, *local_writer->second.writer_handle, *remote_reader_handle,
                                exception)) {
                            if (remote_participant_key == participant_->getGuid()) {
                                logDebug(SECURITY, "Process successful discovering local reader "
                                                       << remote_reader_data.guid() << " unicastlocatorsize:"
                                                       << remote_reader_data.remote_locators().unicast.size());
                                (void)local_writer->second.associated_readers.emplace(
                                    remote_reader_data.guid(),
                                    std::make_tuple(remote_reader_data, remote_reader_handle));

                                // Search local reader.
                                auto local_reader = reader_handles_.find(remote_reader_data.guid());

                                if (local_reader != reader_handles_.end()) {
                                    returned_value = true;
                                    auto remote_writer = local_reader->second.associated_writers.find(writer_guid);

                                    if (remote_writer != local_reader->second.associated_writers.end()) {
                                        if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                                                *local_reader->second.reader_handle,
                                                *std::get<1>(remote_writer->second), local_writer_crypto_tokens,
                                                exception)) {
                                            local_reader_guid = local_reader->first;
                                            writer_data = &(std::get<0>(remote_writer->second));
                                        } else {
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "Cannot set local reader crypto tokens ("
                                                          << remote_reader_data.guid() << ") - (" << exception.what()
                                                          << ")");
                                        }
                                    } else {
                                        // Store in pendings.
                                        (void)remote_writer_pending_messages_.emplace(
                                            std::make_pair(writer_guid, local_reader->first),
                                            std::move(local_writer_crypto_tokens));
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot find local reader (" << remote_reader_data.guid() << ") - ("
                                                                           << exception.what() << ")");
                                }
                            } else {
                                ParticipantGenericMessage message = generate_writer_crypto_token_message(
                                    remote_participant_key, remote_reader_data.guid(), writer_guid,
                                    local_writer_crypto_tokens);
                                logDebug(SECURITY, "Process successful discovering remote reader "
                                                       << remote_reader_data.guid() << " unicastlocatorsize:"
                                                       << remote_reader_data.remote_locators().unicast.size());

                                auto result = local_writer->second.associated_readers.emplace(
                                    remote_reader_data.guid(),
                                    std::make_tuple(remote_reader_data, remote_reader_handle));
                                if (!result.second) {  // key is exist, update
                                    result.first->second = std::make_tuple(remote_reader_data, remote_reader_handle);
                                }
                                lock.unlock();

                                CacheChange_t* change = create_change_for_message(
                                    message, participant_volatile_message_secure_writer_history_);

                                if (change != nullptr) {
                                    // Serialize message
                                    CDRMessage_t aux_msg(0);
                                    aux_msg.wraps = true;
                                    aux_msg.buffer = change->serializedPayload.data;
                                    aux_msg.length = change->serializedPayload.length;
                                    aux_msg.max_size = change->serializedPayload.max_size;

                                    // Serialize encapsulation
                                    (void)CDRMessage::addOctet(&aux_msg, 0);
                                    aux_msg.msg_endian = DEFAULT_ENDIAN;
                                    change->serializedPayload.encapsulation = PL_DEFAULT_ENCAPSULATION;
                                    (void)CDRMessage::addOctet(&aux_msg, DEFAULT_ENCAPSULATION);
                                    (void)CDRMessage::addUInt16(&aux_msg, 0);

                                    if (CDRMessage::addParticipantGenericMessage(&aux_msg, message)) {
                                        change->serializedPayload.length = aux_msg.length;
                                        participant_volatile_message_secure_writer_history_->set_fragments(change);
                                        // Send
                                        if (participant_volatile_message_secure_writer_history_->add_change(change)) {
                                            logDebug(SECURITY, "Process successful discovering remote reader "
                                                                   << remote_reader_data.guid());
                                            returned_value = true;

                                        } else {
                                            (void)participant_volatile_message_secure_writer_history_->release_change(
                                                change);
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "WriterHistory cannot add the CacheChange_t");
                                        }
                                    } else {
                                        (void)participant_volatile_message_secure_writer_history_->release_change(
                                            change);
                                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                  "Cannot serialize ParticipantGenericMessage");
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_NO_DATA,
                                              "WriterHistory cannot retrieve a CacheChange_t");
                                }
                            }
                        } else {
                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                      "Error generating crypto token. (" << exception.what() << ")");
                        }

                        // Here the mutex has to be unlock.
                        if (lock) {
                            lock.unlock();
                        }

                        // If reader was found and setting of crypto tokens works,
                        // then tell core to match reader and writer.
                        if (local_reader_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_writer_with_local_reader_after_security(
                                local_reader_guid, *writer_data);
                        }

                        // If writer was found and setting of crypto tokens works,
                        // then tell core to match writer and reader.
                        if (pairing_cause_pending_message) {
                            (void)participant_->pairing_remote_reader_with_local_writer_after_security(
                                writer_guid, remote_reader_data);
                        }
                    }
                } else {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Crypto plugin fails registering remote reader "
                                  << remote_reader_data.guid() << " of participant " << remote_participant_key);
                }
            } else {
                logDebug(SECURITY, "Storing remote reader << " << remote_reader_data.guid() << " of participant "
                                                               << remote_participant_key << " on pendings");

                remote_reader_pending_discovery_messages_.push_back(
                    std::make_tuple(remote_reader_data, remote_participant_key, writer_guid));

                returned_value = true;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local writer " << writer_guid << std::endl);
        }
    } else if (returned_value) {
        lock.unlock();
        (void)participant_->pairing_remote_reader_with_local_writer_after_security(writer_guid, remote_reader_data);
    }

    return returned_value;
}

bool SecurityManager::discovered_writer(const GUID_t& reader_guid, const GUID_t& remote_participant_key,
                                        WriterProxyData& remote_writer_data,
                                        const EndpointSecurityAttributes& security_attributes) {
    return discovered_writer(reader_guid, remote_participant_key, remote_writer_data, security_attributes, false);
}

void SecurityManager::remove_writer(const GUID_t& reader_guid, const GUID_t& /*remote_participant_key*/,
                                    const GUID_t& remote_writer_guid) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    if (crypto_plugin_ == nullptr) {
        return;
    }

    std::lock_guard<vbs::shared_mutex> _(mutex_);

    auto local_reader = reader_handles_.find(reader_guid);

    if (local_reader != reader_handles_.end()) {
        SecurityException exception;

        auto wit = local_reader->second.associated_writers.find(remote_writer_guid);

        if (wit != local_reader->second.associated_writers.end()) {
            (void)crypto_plugin_->cryptokeyfactory()->unregister_datawriter(std::get<1>(wit->second), exception);
            (void)local_reader->second.associated_writers.erase(wit);
        } else {
            logDebug(SECURITY, "Cannot find remote writer " << remote_writer_guid << std::endl);
        }
    }
}

bool SecurityManager::discovered_builtin_writer(const GUID_t& reader_guid, const GUID_t& remote_participant_key,
                                                WriterProxyData& remote_writer_data,
                                                const EndpointSecurityAttributes& security_attributes) {
    return discovered_writer(reader_guid, remote_participant_key, remote_writer_data, security_attributes, true);
}

bool SecurityManager::discovered_writer(const GUID_t& reader_guid, const GUID_t& remote_participant_key,
                                        WriterProxyData& remote_writer_data,
                                        const EndpointSecurityAttributes& security_attributes, bool is_builtin) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    std::unique_lock<vbs::shared_mutex> lock(mutex_);

    PermissionsHandle* remote_permissions = nullptr;
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto_handle;
    std::shared_ptr<SecretHandle> shared_secret_handle;
    AuthenticationStatus auth_status(AUTHENTICATION_INIT);

    if (!security_attributes.match(remote_writer_data.security_attributes_,
                                   remote_writer_data.plugin_security_attributes_)) {
        return false;
    }

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_permissions = dp_it->second->get_permissions_handle();
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
            shared_secret_handle = dp_it->second->get_shared_secret();
            auth_status = dp_it->second->get_auth_status();
        }
    }

    // assert(access_plugin_ == nullptr || remote_permissions != nullptr);
    // assert(crypto_plugin_ == nullptr || remote_participant_crypto_handle != nullptr);

    bool returned_value = true;
    SecurityException exception;

    if (!is_builtin) {
        if (participant_->security_attributes().allow_unauthenticated_participants &&
            auth_status != AUTHENTICATION_NOT_AVAILABLE && auth_status != AUTHENTICATION_OK &&
            (security_attributes.is_write_protected || security_attributes.is_read_protected)) {
            //!Do not match if read or write protection is enabled for this local endpoint
            return false;
        }

        if (access_plugin_ != nullptr && remote_permissions != nullptr) {
            if ((returned_value = access_plugin_->check_remote_datawriter(*remote_permissions, domain_id_,
                                                                          remote_writer_data, exception)) == false) {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking create remote writer " << remote_writer_data.guid() << " ("
                                                                 << exception.what() << ")");
            }
        }
    }

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        bool is_key_exchange =
            (remote_writer_data.guid().entityId == participant_volatile_message_secure_writer_entity_id);
        auto local_reader = reader_handles_.find(reader_guid);
        returned_value = false;

        if (local_reader != reader_handles_.end()) {
            if ((remote_participant_crypto_handle != nullptr) && (shared_secret_handle != nullptr)) {
                DatawriterCryptoHandle* remote_writer_handle =
                    crypto_plugin_->cryptokeyfactory()->register_matched_remote_datawriter(
                        *local_reader->second.reader_handle, *remote_participant_crypto_handle, *shared_secret_handle,
                        exception);

                if ((remote_writer_handle != nullptr) && (!remote_writer_handle->nil())) {
                    if (is_key_exchange) {
                        logDebug(SECURITY, "Process successful discovering local writer " << remote_writer_data.guid());
                        (void)local_reader->second.associated_writers.emplace(
                            remote_writer_data.guid(), std::make_tuple(remote_writer_data, remote_writer_handle));
                        lock.unlock();
                        (void)participant_->pairing_remote_writer_with_local_reader_after_security(reader_guid,
                                                                                                   remote_writer_data);
                    } else {
                        // Check pending writer crypto messages.
                        auto pending = remote_writer_pending_messages_.find(
                            std::make_pair(remote_writer_data.guid(), reader_guid));
                        bool pairing_cause_pending_message = false;

                        if (pending != remote_writer_pending_messages_.end()) {
                            if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                                    *local_reader->second.reader_handle, *remote_writer_handle, pending->second,
                                    exception)) {
                                pairing_cause_pending_message = true;
                            } else {
                                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                          "Cannot set remote writer crypto tokens ("
                                              << remote_writer_data.guid() << ") - (" << exception.what() << ")");
                            }

                            (void)remote_writer_pending_messages_.erase(pending);
                        }

                        GUID_t local_writer_guid;
                        ReaderProxyData* reader_data = nullptr;

                        // Get local reader crypto tokens.
                        DatareaderCryptoTokenSeq local_reader_crypto_tokens;
                        if (crypto_plugin_->cryptokeyexchange()->create_local_datareader_crypto_tokens(
                                local_reader_crypto_tokens, *local_reader->second.reader_handle, *remote_writer_handle,
                                exception)) {
                            if (remote_participant_key == participant_->getGuid()) {
                                logDebug(SECURITY,
                                         "Process successful discovering local writer " << remote_writer_data.guid());
                                (void)local_reader->second.associated_writers.emplace(
                                    remote_writer_data.guid(),
                                    std::make_tuple(remote_writer_data, remote_writer_handle));

                                // Search local writer.
                                auto local_writer = writer_handles_.find(remote_writer_data.guid());

                                if (local_writer != writer_handles_.end()) {
                                    returned_value = true;
                                    auto remote_reader = local_writer->second.associated_readers.find(reader_guid);

                                    if (remote_reader != local_writer->second.associated_readers.end()) {
                                        if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                                                *local_writer->second.writer_handle,
                                                *std::get<1>(remote_reader->second), local_reader_crypto_tokens,
                                                exception)) {
                                            local_writer_guid = local_writer->first;
                                            reader_data = &(std::get<0>(remote_reader->second));
                                        } else {
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "Cannot set local writer crypto tokens ("
                                                          << remote_writer_data.guid() << ") - (" << exception.what()
                                                          << ")");
                                        }
                                    } else {
                                        // Store in pendings.
                                        (void)remote_reader_pending_messages_.emplace(
                                            std::make_pair(reader_guid, local_writer->first),
                                            std::move(local_reader_crypto_tokens));
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot find local writer (" << remote_writer_data.guid() << ") - ("
                                                                           << exception.what() << ")");
                                }
                            } else {
                                ParticipantGenericMessage message = generate_reader_crypto_token_message(
                                    remote_participant_key, remote_writer_data.guid(), reader_guid,
                                    local_reader_crypto_tokens);
                                auto result = local_reader->second.associated_writers.emplace(
                                    remote_writer_data.guid(),
                                    std::make_tuple(remote_writer_data, remote_writer_handle));
                                if (!result.second) {  // key is exist, update
                                    result.first->second = std::make_tuple(remote_writer_data, remote_writer_handle);
                                }
                                lock.unlock();

                                CacheChange_t* change = create_change_for_message(
                                    message, participant_volatile_message_secure_writer_history_);

                                if (change != nullptr) {
                                    // Serialize message
                                    CDRMessage_t aux_msg(0);
                                    aux_msg.wraps = true;
                                    aux_msg.buffer = change->serializedPayload.data;
                                    aux_msg.length = change->serializedPayload.length;
                                    aux_msg.max_size = change->serializedPayload.max_size;

                                    // Serialize encapsulation
                                    (void)CDRMessage::addOctet(&aux_msg, 0);
                                    aux_msg.msg_endian = DEFAULT_ENDIAN;
                                    change->serializedPayload.encapsulation = PL_DEFAULT_ENCAPSULATION;
                                    (void)CDRMessage::addOctet(&aux_msg, DEFAULT_ENCAPSULATION);
                                    (void)CDRMessage::addUInt16(&aux_msg, 0);

                                    if (CDRMessage::addParticipantGenericMessage(&aux_msg, message)) {
                                        change->serializedPayload.length = aux_msg.length;

                                        // Send
                                        participant_volatile_message_secure_writer_history_->set_fragments(change);
                                        if (participant_volatile_message_secure_writer_history_->add_change(change)) {
                                            logDebug(SECURITY, "Process successful discovering remote writer "
                                                                   << remote_writer_data.guid());
                                            returned_value = true;
                                        } else {
                                            (void)participant_volatile_message_secure_writer_history_->release_change(
                                                change);
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "WriterHistory cannot add the CacheChange_t");
                                        }
                                    } else {
                                        (void)participant_volatile_message_secure_writer_history_->release_change(
                                            change);
                                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                  "Cannot serialize ParticipantGenericMessage");
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_NO_DATA,
                                              "WriterHistory cannot retrieve a CacheChange_t");
                                }
                            }
                        } else {
                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                      "Error generating crypto token. (" << exception.what() << ")");
                        }

                        // Here the mutex has to be unlock.
                        if (lock) {
                            lock.unlock();
                        }

                        // If writer was found and setting of crypto tokens works,
                        // then tell core to match writer and reader.
                        if (local_writer_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_reader_with_local_writer_after_security(
                                local_writer_guid, *reader_data);
                        }

                        // If reader was found and setting of crypto tokens works,
                        // then tell core to match reader and writer.
                        if (pairing_cause_pending_message) {
                            (void)participant_->pairing_remote_writer_with_local_reader_after_security(
                                reader_guid, remote_writer_data);
                        }
                    }
                } else {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Crypto plugin fails registering remote writer "
                                  << remote_writer_data.guid() << " of participant " << remote_participant_key);
                }
            } else {
                logDebug(SECURITY, "Storing remote writer << " << remote_writer_data.guid() << " of participant "
                                                               << remote_participant_key << "on pendings");

                remote_writer_pending_discovery_messages_.push_back(
                    std::make_tuple(remote_writer_data, remote_participant_key, reader_guid));

                returned_value = true;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local reader " << reader_guid << std::endl);
        }
    } else if (returned_value) {
        lock.unlock();
        (void)participant_->pairing_remote_writer_with_local_reader_after_security(reader_guid, remote_writer_data);
    }

    return returned_value;
}

bool SecurityManager::encode_writer_submessage(const CDRMessage_t& input_message, CDRMessage_t& output_message,
                                               const GUID_t& writer_guid,
                                               const std::vector<GUID_t>& receiving_list) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    if (writer_guid.entityId == participant_volatile_message_secure_writer_entity_id) {
        bool ret_val = false;
        if (receiving_list.size() == 1) {
            GUID_t participant_guid(receiving_list.at(0).guidPrefix, c_EntityId_RTPSParticipant);
            auto part = discovered_participants_.find(participant_guid);
            if (part != discovered_participants_.end()) {
                auto pCrypto = part->second->get_participant_crypto();

                if (pCrypto) {
                    EndpointSecurityAttributes attr;
                    SecurityException exception;
                    PropertySeq auxProps;
                    auxProps.emplace_back(
                        Property("dds.sec.builtin_endpoint_name", "BuiltinParticipantVolatileMessageSecureWriter"));
                    auto wHandle = crypto_plugin_->cryptokeyfactory()->register_local_datawriter(*pCrypto, auxProps,
                                                                                                 attr, exception);
                    std::vector<std::shared_ptr<DatareaderCryptoHandle>> receiving_crypto_list;
                    if (wHandle != nullptr) {
                        ret_val = crypto_plugin_->cryptotransform()->encode_datawriter_submessage(
                            output_message, input_message, *wHandle, receiving_crypto_list, exception);
                    }
                }
            }
        }

        return ret_val;
    }

    const auto& wr_it = writer_handles_.find(writer_guid);

    if (wr_it != writer_handles_.end()) {
        std::vector<std::shared_ptr<DatareaderCryptoHandle>> receiving_datareader_crypto_list;

        for (const auto& rd_it : receiving_list) {
            const auto rd_it_handle = wr_it->second.associated_readers.find(rd_it);

            if (rd_it_handle != wr_it->second.associated_readers.end()) {
                try {
                    receiving_datareader_crypto_list.emplace_back(
                        std::get<1>(rd_it_handle->second)->shared_from_this());
                } catch (const std::bad_weak_ptr&) {
                    elogError(SECURITY, RetCode_t::RETCODE_BAD_PARAMETER, "throw except std::bad_weak_ptr");
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find remote reader " << rd_it);
            }
        }

        if (receiving_datareader_crypto_list.size() > 0) {
            SecurityException exception;

            if (crypto_plugin_->cryptotransform()->encode_datawriter_submessage(
                    output_message, input_message, *wr_it->second.writer_handle, receiving_datareader_crypto_list,
                    exception)) {
                return true;
            }
        }
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local writer " << writer_guid);
    }

    return false;
}

bool SecurityManager::encode_reader_submessage(const CDRMessage_t& input_message, CDRMessage_t& output_message,
                                               const GUID_t& reader_guid,
                                               const std::vector<GUID_t>& receiving_list) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    if (reader_guid.entityId == participant_volatile_message_secure_reader_entity_id) {
        bool ret_val = false;

        if (receiving_list.size() == 1) {
            GUID_t participant_guid(receiving_list.at(0).guidPrefix, c_EntityId_RTPSParticipant);
            auto part = discovered_participants_.find(participant_guid);
            if (part != discovered_participants_.end()) {
                auto pCrypto = part->second->get_participant_crypto();

                if (pCrypto) {
                    EndpointSecurityAttributes attr;
                    SecurityException exception;
                    PropertySeq auxProps;
                    auxProps.emplace_back(
                        Property("dds.sec.builtin_endpoint_name", "BuiltinParticipantVolatileMessageSecureReader"));
                    auto rHandle = crypto_plugin_->cryptokeyfactory()->register_local_datareader(*pCrypto, auxProps,
                                                                                                 attr, exception);
                    std::vector<std::shared_ptr<DatawriterCryptoHandle>> receiving_crypto_list;
                    if (rHandle != nullptr) {
                        ret_val = crypto_plugin_->cryptotransform()->encode_datareader_submessage(
                            output_message, input_message, *rHandle, receiving_crypto_list, exception);
                    }
                }
            }
        }

        return ret_val;
    }

    const auto& rd_it = reader_handles_.find(reader_guid);

    if (rd_it != reader_handles_.end()) {
        std::vector<std::shared_ptr<DatawriterCryptoHandle>> receiving_datawriter_crypto_list;

        for (const auto& wr_it : receiving_list) {
            const auto wr_it_handle = rd_it->second.associated_writers.find(wr_it);

            if (wr_it_handle != rd_it->second.associated_writers.end()) {
                try {
                    receiving_datawriter_crypto_list.push_back(std::get<1>(wr_it_handle->second)->shared_from_this());
                } catch (const std::bad_weak_ptr& e) {
                    (void)e;
                    elogError(SECURITY, RetCode_t::RETCODE_BAD_PARAMETER,
                              std::string("catch std::bad_weak_ptr exception:") << e.what());
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find remote writer " << wr_it);
            }
        }

        if (receiving_datawriter_crypto_list.size() > 0) {
            SecurityException exception;

            if (crypto_plugin_->cryptotransform()->encode_datareader_submessage(
                    output_message, input_message, *rd_it->second.reader_handle, receiving_datawriter_crypto_list,
                    exception)) {
                return true;
            }
        }
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local reader " << reader_guid);
    }

    return false;
}

int SecurityManager::decode_rtps_submessage(CDRMessage_t& message, CDRMessage_t& out_message,
                                            const GuidPrefix_t& sending_participant) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return -1;
    }
    if (message.buffer[message.pos] != SEC_PREFIX) {
        return 1;
    }

    if (crypto_plugin_ == nullptr) {
        return 0;
    }

    (void)CDRMessage::initCDRMsg(&out_message);

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    const GUID_t remote_participant_key(sending_participant, c_EntityId_RTPSParticipant);
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto_handle;

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
        }
    }

    if (remote_participant_crypto_handle != nullptr) {
        DatawriterCryptoHandle* writer_handle = nullptr;
        DatareaderCryptoHandle* reader_handle = nullptr;
        SecureSubmessageCategory_t category = INFO_SUBMESSAGE;
        SecurityException exception;

        // encoding requires modifying the handle but it takes care of its own synchronization
        if (crypto_plugin_->cryptotransform()->preprocess_secure_submsg(&writer_handle, &reader_handle, category,
                                                                        message, *local_participant_crypto_handle_,
                                                                        *remote_participant_crypto_handle, exception)) {
            // TODO (Ricardo) Category INFO
            if (category == DATAWRITER_SUBMESSAGE) {
                if (crypto_plugin_->cryptotransform()->decode_datawriter_submessage(
                        out_message, message, *reader_handle, *writer_handle, exception)) {
                    return 0;
                } else {
                    logWarning(SECURITY, "Cannot decode writer RTPS submessage (" << exception.what() << ")");
                }
            } else if (category == DATAREADER_SUBMESSAGE) {
                if (crypto_plugin_->cryptotransform()->decode_datareader_submessage(
                        out_message, message, *writer_handle, *reader_handle, exception)) {
                    return 0;
                } else {
                    logWarning(SECURITY, "Cannot decode reader RTPS submessage (" << exception.what() << ")");
                }
            }
        } else {
            logDebug(SECURITY, "Cannot preprocess RTPS submessage (" << exception.what() << ")");
        }
    } else {
        logDebug(SECURITY, "Cannot decode RTPS submessage for participant " << remote_participant_key);
    }

    return -1;
}

bool SecurityManager::encode_serialized_payload(const SerializedPayload_t& payload, SerializedPayload_t& output_payload,
                                                const GUID_t& writer_guid) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    const auto& wr_it = writer_handles_.find(writer_guid);

    if (wr_it != writer_handles_.end()) {
        SecurityException exception;

        if (crypto_plugin_->cryptotransform()->encode_serialized_payload(output_payload, payload,
                                                                         *wr_it->second.writer_handle, exception)) {
            return true;
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Error encoding payload failed");
        }
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local writer " << writer_guid);
    }

    return false;
}

bool SecurityManager::decode_serialized_payload(const SerializedPayload_t& secure_payload, SerializedPayload_t& payload,
                                                const GUID_t& reader_guid, const GUID_t& writer_guid) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    if (crypto_plugin_ == nullptr) {
        return false;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    const auto& rd_it = reader_handles_.find(reader_guid);

    if (rd_it != reader_handles_.end()) {
        const auto wr_it_handle = rd_it->second.associated_writers.find(writer_guid);

        if (wr_it_handle != rd_it->second.associated_writers.end()) {
            SecurityException exception;

            if (crypto_plugin_->cryptotransform()->decode_serialized_payload(
                    payload, secure_payload, *rd_it->second.reader_handle, *std::get<1>(wr_it_handle->second),
                    exception)) {
                return true;
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error decoding encoded payload (" << exception.what() << ")");
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find remote writer " << writer_guid);
        }
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local reader " << reader_guid);
    }

    return false;
}

bool SecurityManager::participant_authorized(const ParticipantProxyData& participant_data,
                                             const DiscoveredParticipantInfo::AuthUniquePtr& remote_participant_info,
                                             std::shared_ptr<SecretHandle>& shared_secret_handle) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    logInfo(SECURITY, "Authorized participant " << participant_data.m_guid);

    SecurityException exception;
    PermissionsHandle* remote_permissions = nullptr;

    if (access_plugin_ != nullptr) {  //&& !get_mvbs_status(*remote_participant_info->identity_handle_)
        PermissionsCredentialToken* credential_token = nullptr;
        if (authentication_plugin_->get_authenticated_peer_credential_token(
                &credential_token, *remote_participant_info->identity_handle_, exception)) {
            remote_permissions = access_plugin_->validate_remote_permissions(
                *authentication_plugin_, *local_identity_handle_, *local_permissions_handle_,
                *remote_participant_info->identity_handle_, participant_data.permissions_token_, *credential_token,
                exception);

            if ((remote_permissions != nullptr) && (!remote_permissions->nil())) {
                if (!access_plugin_->check_remote_participant(*remote_permissions, domain_id_, participant_data,
                                                              exception)) {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Error checking remote participant  " << participant_data.m_guid << " ("
                                                                    << exception.what() << ").");
                    (void)access_plugin_->return_permissions_handle(remote_permissions, exception);
                    remote_permissions = nullptr;
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error validating remote permissions for " << participant_data.m_guid << " ("
                                                                     << exception.what() << ").");

                if (remote_permissions != nullptr) {
                    (void)access_plugin_->return_permissions_handle(remote_permissions, exception);
                    remote_permissions = nullptr;
                }
            }

            (void)authentication_plugin_->return_authenticated_peer_credential_token(credential_token, exception);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                      "Not receive remote permissions of participant " << participant_data.m_guid << " ("
                                                                       << exception.what() << ").");
        }
    }

    if ((access_plugin_ == nullptr) || (remote_permissions != nullptr) ||
        get_mvbs_status(*remote_participant_info->identity_handle_)) {
        std::list<std::pair<ReaderProxyData, GUID_t>> temp_readers;
        std::list<std::pair<WriterProxyData, GUID_t>> temp_writers;

        if (crypto_plugin_ != nullptr && !get_mvbs_status(*remote_participant_info->identity_handle_)) {
            // TODO(Ricardo) Study cryptography without sharedsecret
            if (!shared_secret_handle) {
                elogError(SECURITY, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                          "Not shared secret for participant " << participant_data.m_guid);
                return false;
            }

            // Starts cryptography mechanism
            std::shared_ptr<ParticipantCryptoHandle> participant_crypto_handle =
                register_and_match_crypto_endpoint(*remote_participant_info->identity_handle_, *shared_secret_handle);

            // Store cryptography info
            if (participant_crypto_handle && (!participant_crypto_handle->nil())) {
                std::lock_guard<vbs::shared_mutex> _(mutex_);

                // Check there is a pending crypto message.
                auto pending = remote_participant_pending_messages_.find(participant_data.m_guid);

                if (pending != remote_participant_pending_messages_.end()) {
                    if (!crypto_plugin_->cryptokeyexchange()->set_remote_participant_crypto_tokens(
                            *local_participant_crypto_handle_, *participant_crypto_handle, pending->second,
                            exception)) {
                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                  "Cannot set remote participant crypto tokens (" << participant_data.m_guid << ") - ("
                                                                                  << exception.what() << ")");
                    }

                    (void)remote_participant_pending_messages_.erase(pending);
                }

                // Search in pendings readers and writers
                auto rit = remote_reader_pending_discovery_messages_.begin();
                while (rit != remote_reader_pending_discovery_messages_.end()) {
                    if (std::get<1>(*rit) == participant_data.m_guid) {
                        temp_readers.push_back(std::make_pair(std::get<0>(*rit), std::get<2>(*rit)));
                        rit = remote_reader_pending_discovery_messages_.erase(rit);
                        continue;
                    }

                    ++rit;
                }

                auto wit = remote_writer_pending_discovery_messages_.begin();
                while (wit != remote_writer_pending_discovery_messages_.end()) {
                    if (std::get<1>(*wit) == participant_data.m_guid) {
                        temp_writers.push_back(std::make_pair(std::get<0>(*wit), std::get<2>(*wit)));
                        wit = remote_writer_pending_discovery_messages_.erase(wit);
                        continue;
                    }

                    ++wit;
                }

                auto dp_it = discovered_participants_.find(participant_data.m_guid);

                if (dp_it != discovered_participants_.end()) {
                    dp_it->second->set_participant_crypto(participant_crypto_handle);
                    dp_it->second->set_shared_secret(shared_secret_handle);
                    dp_it->second->set_permissions_handle(remote_permissions);
                } else {
                    (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(participant_crypto_handle,
                                                                                     exception);
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Cannot find remote participant " << participant_data.m_guid);
                    return false;
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Cannot register remote participant in crypto plugin (" << participant_data.m_guid << ")");
                return false;
            }

            if (!get_mvbs_status(*remote_participant_info->identity_handle_))
                match_builtin_key_exchange_endpoints(participant_data);
            exchange_participant_crypto(std::move(participant_crypto_handle), participant_data.m_guid);
        } else {
            {
                vbs::shared_lock<vbs::shared_mutex> _(mutex_);

                // Store shared_secret.
                auto dp_it = discovered_participants_.find(participant_data.m_guid);

                if (dp_it != discovered_participants_.end()) {
                    dp_it->second->set_shared_secret(shared_secret_handle);
                    dp_it->second->set_permissions_handle(remote_permissions);
                }
            }
            if (!get_mvbs_status(*remote_participant_info->identity_handle_))
                match_builtin_key_exchange_endpoints(participant_data);
        }

        logDebug(SECURITY, "Participant " << participant_data.m_guid << " authenticated");

        if (!get_mvbs_status(*remote_participant_info->identity_handle_)) {
            for (auto& remote_reader : temp_readers) {
                (void)participant_->pdp()->getEDP()->pairing_reader_proxy_with_local_writer(
                    remote_reader.second, participant_data.m_guid, remote_reader.first);
            }

            for (auto& remote_writer : temp_writers) {
                (void)participant_->pdp()->getEDP()->pairing_writer_proxy_with_local_reader(
                    remote_writer.second, participant_data.m_guid, remote_writer.first);
            }
        }
        return true;
    }

    return false;
}

void SecurityManager::notify_participant_authorized(const ParticipantProxyData& participant_data) {
    std::string part_type = participant_->pdp()->check_participant_type(participant_data.m_properties);
    if (participant_->getAttributes().builtin.discovery_config.discoveryProtocol == DiscoveryProtocol_t::SIMPLE &&
        (part_type == edds::rtps::ParticipantType::SERVER || part_type == edds::rtps::ParticipantType::BACKUP)) {
        SecurityException exception;
        // just like on_validation_failed and allow_unauthenticated process
        participant_->pdp()->notifyAboveRemoteEndpoints(participant_data, false);

        if (strlen(exception.what()) > 0) {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, exception.what());
        }

        logInfo(SECURITY, "Authentication success for simple participant " << participant_data.m_guid);

        // Inform user about authenticated remote participant.
        vbs::common::ParticipantListener* listener = participant_->getListener();
        if (participant_->getListener() != nullptr) {
            vbs::ParticipantAuthenticationInfo info;
            info.status = vbs::ParticipantAuthenticationInfo::AUTHORIZED_PARTICIPANT;
            info.guid = participant_data.m_guid;
            listener->onParticipantAuthentication(
                dynamic_cast<vbs::dispatcher::SubstrateParticipant*>(participant_->getUserRTPSParticipant()),
                std::move(info));
        }
        return;
    }

    participant_->pdp()->notifyAboveRemoteEndpoints(participant_data, true);

    logInfo(SECURITY, "Participant " << participant_data.m_guid << " authenticated");

    // Inform user about authenticated remote participant.
    vbs::common::ParticipantListener* listener = participant_->getListener();
    if (listener != nullptr) {
        vbs::ParticipantAuthenticationInfo info;
        info.status = vbs::ParticipantAuthenticationInfo::AUTHORIZED_PARTICIPANT;
        info.guid = participant_data.m_guid;
        listener->onParticipantAuthentication(
            dynamic_cast<vbs::dispatcher::SubstrateParticipant*>(participant_->getUserRTPSParticipant()),
            std::move(info));
    }
}

uint32_t SecurityManager::calculate_extra_size_for_rtps_message() const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return 0;
    }

    if (crypto_plugin_ == nullptr) {
        return 0;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    return crypto_plugin_->cryptotransform()->calculate_extra_size_for_rtps_message(
        static_cast<uint32_t>(discovered_participants_.size()));
}

uint32_t SecurityManager::calculate_extra_size_for_rtps_submessage(const GUID_t& writer_guid) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return 0;
    }

    if (crypto_plugin_ == nullptr) {
        return 0;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    auto wr_it = writer_handles_.find(writer_guid);

    if (wr_it != writer_handles_.end()) {
        return crypto_plugin_->cryptotransform()->calculate_extra_size_for_rtps_submessage(
            static_cast<uint32_t>(wr_it->second.associated_readers.size()));
    } else {
        logDebug(SECURITY, "Not found local writer " << writer_guid);
    }

    return 0;
}

uint32_t SecurityManager::calculate_extra_size_for_encoded_payload(const GUID_t& writer_guid) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return 0;
    }

    if (crypto_plugin_ == nullptr) {
        return 0;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    auto wr_it = writer_handles_.find(writer_guid);

    if (wr_it != writer_handles_.end()) {
        return crypto_plugin_->cryptotransform()->calculate_extra_size_for_encoded_payload(
            static_cast<uint32_t>(wr_it->second.associated_readers.size()));
    } else {
        logDebug(SECURITY, "Not found local writer " << writer_guid);
    }

    return 0;
}

void SecurityManager::resend_handshake_message_token(const GUID_t& remote_participant_key) const {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return;
    }

    vbs::shared_lock<vbs::shared_mutex> _(mutex_);

    auto dp_it = discovered_participants_.find(remote_participant_key);

    if (dp_it != discovered_participants_.end()) {
        SecurityManager::DiscoveredParticipantInfo::AuthUniquePtr remote_participant_info = dp_it->second->get_auth();

        if (remote_participant_info) {

            if (remote_participant_info->handshake_requests_sent_ >= auth_handshake_props_.max_handshake_requests_) {
                if (remote_participant_info->auth_status_ != AUTHENTICATION_FAILED) {
                    SecurityException exception;
                    remote_participant_info->event_->cancel_timer();
                    remote_participant_info->auth_status_ = AUTHENTICATION_FAILED;
                    on_validation_failed(dp_it->second->participant_data(), exception);
                }
            } else {
                if (remote_participant_info->change_sequence_number_ != SequenceNumber_t::unknown()) {
                    std::lock_guard<std::mutex> lck(mtx);
                    CacheChange_t* p_change = participant_stateless_message_writer_history_->remove_change_and_reuse(
                        remote_participant_info->change_sequence_number_);
                    remote_participant_info->change_sequence_number_ = SequenceNumber_t::unknown();

                    if (p_change != nullptr) {
                        logInfo(SECURITY, "Authentication handshake resent to participant "
                                              << remote_participant_key
                                              << "my auth:" << remote_participant_info->auth_status_);
                        participant_stateless_message_writer_history_->set_fragments(p_change);

                        if (participant_stateless_message_writer_history_->add_change(p_change)) {
                            remote_participant_info->change_sequence_number_ = p_change->sequenceNumber;
                            remote_participant_info->handshake_requests_sent_++;
                        }
                        // TODO (Ricardo) What to do if not added?
                    }
                }

                if (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REPLY) {
                    // Avoid DoS attack by exponentially increasing event interval
                    auto time_ms = remote_participant_info->event_->getIntervalMilliSec();
                    remote_participant_info->event_->update_interval_millisec(
                        time_ms * auth_handshake_props_.handshake_resend_period_gain_);
                    remote_participant_info->event_->restart_timer();
                }
            }

            dp_it->second->set_auth(remote_participant_info);
        }
    }
}

bool SecurityManager::get_mvbs_status(const IdentityHandle& remote_identity_handle) {
    const PKIIdentityHandle& rih = PKIIdentityHandle::narrow(remote_identity_handle);

    if (rih.nil()) {
        return false;
    }

    return rih->ismvbs_sec;
}

void SecurityManager::on_validation_failed(const ParticipantProxyData& participant_data,
                                           const SecurityException& exception) const {
    if (participant_->security_attributes().allow_unauthenticated_participants) {
        participant_->pdp()->notifyAboveRemoteEndpoints(participant_data, false);
    }

    if (strlen(exception.what()) > 0) {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, exception.what());
    }

    logInfo(SECURITY, "Authentication failed for participant " << participant_data.m_guid);

    // Inform user about authenticated remote participant.
    vbs::common::ParticipantListener* listener = participant_->getListener();
    if (listener != nullptr) {
        vbs::ParticipantAuthenticationInfo info;
        info.status = vbs::ParticipantAuthenticationInfo::UNAUTHORIZED_PARTICIPANT;
        info.guid = participant_data.m_guid;
        listener->onParticipantAuthentication(
            dynamic_cast<vbs::dispatcher::SubstrateParticipant*>(participant_->getUserRTPSParticipant()),
            std::move(info));
    }
}

SecurityManager::AuthenticationHandshakeProperties::AuthenticationHandshakeProperties()
    : max_handshake_requests_(100), initial_handshake_resend_period_ms_(125), handshake_resend_period_gain_(1.5) {}

void SecurityManager::AuthenticationHandshakeProperties::parse_from_property_policy(
    const PropertyPolicy& auth_handshake_properties) {
    const Property* const max_handshake_requests =
        PropertyPolicyHelper::get_property(auth_handshake_properties, "max_handshake_requests");

    if (max_handshake_requests != nullptr) {
        max_handshake_requests_ =
            (int32_t)PropertyParser::as_int(*max_handshake_requests, false, 0, true, 1,
                                            SecurityException("Error parsing max_handshake_requests property value."));
    }

    const Property* const initial_handshake_resend_period =
        PropertyPolicyHelper::get_property(auth_handshake_properties, "initial_handshake_resend_period");

    if (initial_handshake_resend_period != nullptr) {
        initial_handshake_resend_period_ms_ = (int32_t)PropertyParser::as_int(
            *initial_handshake_resend_period, false, 0, true, 1,
            SecurityException("Error parsing initial_handshake_resend_period property value."));
    }

    const Property* const handshake_resend_period_gain =
        PropertyPolicyHelper::get_property(auth_handshake_properties, "handshake_resend_period_gain");

    if (handshake_resend_period_gain != nullptr) {
        handshake_resend_period_gain_ =
            PropertyParser::as_double(*handshake_resend_period_gain, false, 0.0, true, 1.0,
                                      SecurityException("Error parsing handshake_resend_period_gain property value."));
    }
}

bool SecurityManager::DiscoveredParticipantInfo::check_guid_comes_from(Authentication* const auth_plugin,
                                                                       const GUID_t& adjusted, const GUID_t& original) {
    bool ret = false;
    if (auth_plugin != nullptr) {
        std::lock_guard<std::mutex> g(mtx_);

        if (nullptr != auth_ && AuthenticationStatus::AUTHENTICATION_OK == auth_->auth_status_) {
            ret = auth_plugin->check_guid_comes_from(auth_->identity_handle_, adjusted, original);
        }
    }
    return ret;
}

void SecurityManager::onWriterChangeReceivedByAll(vbs::dispatcher::SubstrateWriter* writer, CacheChange_t* change) {
    static_cast<void>(writer);
    assert(writer == participant_volatile_message_secure_writer_);

    if (nullptr != participant_volatile_message_secure_writer_history_) {
        participant_volatile_message_secure_writer_history_->remove_change(change);
    }
}

std::pair<uint32_t, uint32_t> SecurityManager::compute_authtoken(const GuidPrefix_t& src, const GuidPrefix_t& dst,
                                                                 uint64_t seq, security::AuthToken& token,
                                                                 const std::vector<uint8_t>& key, bool psk) {
    uint32_t guid_size = GuidPrefix_t::size + EntityId_t::size;
    std::vector<uint8_t> data(guid_size * 2, 0);

    // {little guid | big guid}
    if (src > dst) {
        std::memcpy(data.data(), dst.value, GuidPrefix_t::size);
        std::memcpy(data.data() + guid_size, src.value, GuidPrefix_t::size);
    } else {
        std::memcpy(data.data(), src.value, GuidPrefix_t::size);
        std::memcpy(data.data() + guid_size, dst.value, GuidPrefix_t::size);
    }

    uint32_t id = token.max_blocks_per_session ? ((seq - 1) / token.max_blocks_per_session) : 0;

    if (psk) {
        return token.compute_authtoken(data, id);
    } else {
        return token.compute_authtoken(key, data, id);
    }
}

void SecurityManager::insert_local_token(const GuidPrefix_t& key, uint32_t expected_token) {
    std::lock_guard<vbs::shared_mutex> lock(local_token_mtx_);
    auto it = local_auth_tokens_.find(key);
    if (it == local_auth_tokens_.end()) {
        security::AuthToken& token = local_auth_tokens_[key];
        token.firstToken() = expected_token;
        token.authToken(expected_token);
    }
}

void SecurityManager::insert_remote_token(const GuidPrefix_t& key, uint32_t expected_token) {
    std::lock_guard<vbs::shared_mutex> lock(remote_token_mtx_);
    auto it = remote_auth_tokens_.find(key);
    if (it == remote_auth_tokens_.end()) {
        security::AuthToken& token = remote_auth_tokens_[key];
        token.firstToken() = expected_token;
        token.authToken(expected_token);
    }
}

uint32_t SecurityManager::get_authtoken(const GuidPrefix_t& src, const GuidPrefix_t& dst, uint64_t seq) {
    uint32_t m_token = 0;
    const std::vector<uint8_t> dummy_key(1, 0);
    const std::vector<uint8_t>* key = nullptr;
    std::pair<uint32_t, uint32_t> expected {0, 0};
    uint32_t remote_token = 0;

    VendorId_t vid;
    dst.get_vendor_id(vid);
    if (vid != c_VendorId_vbs && vid != c_VendorId_mvbs) {
        logDebug(AUTHTOKEN, "vendorid is not vbs or mvbs:" << dst);
        return m_token;
    }

    if (vid == c_VendorId_vbs) {
        key = get_shared_key(dst);
        if (key == nullptr) {
            return m_token;
        }
    }

    {
        std::lock_guard<vbs::shared_mutex> _(local_token_mtx_);
        auto it = local_auth_tokens_.find(dst);
        // Get token and update if necessary
        // If the maximum number of blocks have been processed, generate a new token
        if (it == local_auth_tokens_.end() ||
            (it->second.max_blocks_per_session != 0 && seq > it->second.session_block_edge)) {
            security::AuthToken& token = (it == local_auth_tokens_.end()) ? local_auth_tokens_[dst] : it->second;

            expected = compute_authtoken(src, dst, seq, token, (key ? *key : dummy_key), vid == c_VendorId_mvbs);
            if (src > dst) {
                token.firstToken() = expected.second;
                token.authToken(expected.second);
                remote_token = expected.first;
            } else {
                token.firstToken() = expected.first;
                token.authToken(expected.first);
                remote_token = (src == dst ? expected.first : expected.second);
            }
            m_token = token.authToken();

            // In any case, increment session block counter
            if (it != local_auth_tokens_.end()) {
                token.session_block_edge += token.max_blocks_per_session;
            }
        } else {
            m_token = it->second.authToken();
            logDebug(AUTHTOKEN, "find authtoken:" << src << "->" << dst);
        }
    }

    insert_remote_token(dst, remote_token);

    return m_token;
}

bool SecurityManager::checkAuthToken(const GUID_t& fromGUID, const GUID_t& toGUID, uint64_t seq, uint32_t auth_token) {
    const GuidPrefix_t& src = fromGUID.guidPrefix;
    GuidPrefix_t dst;
    const std::vector<uint8_t> dummy_key(1, 0);
    const std::vector<uint8_t>* key = nullptr;
    std::pair<uint32_t, uint32_t> expected {0, 0};
    uint32_t local_token = 0;

    if (auth_token == 0) {
        logDebug(AUTHTOKEN, "authtoken can't be zero " << fromGUID << "->" << toGUID);
        return false;
    }

    VendorId_t vid;
    src.get_vendor_id(vid);
    if (vid != c_VendorId_vbs && vid != c_VendorId_mvbs) {
        return true;
    }

    if (vid == c_VendorId_vbs) {
        key = get_shared_key(src);
        if (key == nullptr) {
            return false;
        }
    }

    if (toGUID.guidPrefix == c_GuidPrefix_Unknown) {
        dst = participant_->getGuid().guidPrefix;
    } else {
        dst = toGUID.guidPrefix;
    }

    {
        std::lock_guard<vbs::shared_mutex> _(remote_token_mtx_);
        auto it = remote_auth_tokens_.find(src);
        // Check token and update if necessary
        if (it == remote_auth_tokens_.end() || auth_token != it->second.authToken()) {
            security::AuthToken& token = (it == remote_auth_tokens_.end()) ? remote_auth_tokens_[src] : it->second;

            if (it != remote_auth_tokens_.end() && token.max_blocks_per_session == 0) {
                logDebug(AUTHTOKEN, "Auth token check error " << fromGUID << "->" << toGUID);
                return false;
            }

            expected = compute_authtoken(src, dst, seq, token, (key ? *key : dummy_key), vid == c_VendorId_mvbs);
            if (src > dst) {
                token.firstToken() = expected.second;
                token.authToken(expected.second);
                local_token = expected.first;
            } else {
                token.firstToken() = expected.first;
                token.authToken(expected.first);
                local_token = (src == dst ? expected.first : expected.second);
            }

            if (auth_token != token.authToken()) {
                logDebug(AUTHTOKEN, "Auth token check error " << fromGUID << "->" << toGUID);
                return false;
            }
        } else {
            logDebug(AUTHTOKEN, "Auth token check OK " << fromGUID << "->" << toGUID);
        }
    }

    insert_local_token(src, local_token);

    return true;
}

void SecurityManager::removeRemoteAuthToken(const GUID_t& to_remove) const {
    std::lock_guard<vbs::shared_mutex> _(remote_token_mtx_);
    remote_auth_tokens_.erase(to_remove.guidPrefix);
}

bool SecurityManager::discovered_participant_without_handshake(const ParticipantProxyData& participant_data) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }
    is_client = true;

    // Early return when ParticipantSecurityInfo does not match
    auto& sec_attrs = participant_->security_attributes();
    if (!sec_attrs.match(participant_data.security_attributes_, participant_data.plugin_security_attributes_)) {
        logInfo(SECURITY, "participant security_attributes unmatch with plugin: "
                              << participant_data.security_attributes_ << " "
                              << participant_data.plugin_security_attributes_);
        return false;
    }

    if (authentication_plugin_ == nullptr) {
        participant_->pdp()->notifyAboveRemoteEndpoints(participant_data, true);
        return true;
    }

    SecurityException exception;
    AuthenticationStatus auth_status = AUTHENTICATION_INIT;

    // Create or find information
    bool undiscovered = false;
    DiscoveredParticipantInfo::AuthUniquePtr remote_participant_info;

    // Use the information from the collection
    ParticipantProxyData remote_participant_data;
    {
        std::lock_guard<vbs::shared_mutex> _(mutex_);
        auto map_ret = discovered_participants_.insert(std::make_pair(
            participant_data.m_guid,
            std::unique_ptr<DiscoveredParticipantInfo>(new DiscoveredParticipantInfo(auth_status, participant_data))));

        undiscovered = map_ret.second;
        remote_participant_info = map_ret.first->second->get_auth();
        remote_participant_data = map_ret.first->second->participant_data();
        logDebug(SECURITY, "remote participant_data is client:"
                               << participant_data.is_client << " remote_participant_data is client:"
                               << remote_participant_data.is_client << " new insert new:" << undiscovered
                               << " participant_data():" << map_ret.first->second->participant_data().is_client);
    }

    bool notify_part_authorized = false;
    if (undiscovered && remote_participant_info) {
        IdentityHandle* remote_identity_handle = nullptr;

        ValidationResult_t validation_ret = authentication_plugin_->validate_remote_identity(
            &remote_identity_handle, *local_identity_handle_, remote_participant_data.identity_token_,
            remote_participant_data.m_guid, exception);
        switch (validation_ret) {
            case VALIDATION_OK:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_OK;
                break;
            case VALIDATION_PENDING_HANDSHAKE_REQUEST:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_REQUEST_NOT_SEND;
                break;
            case VALIDATION_PENDING_HANDSHAKE_MESSAGE:
                assert(remote_identity_handle != nullptr);
                auth_status = AUTHENTICATION_WAITING_REQUEST;
                break;
            case VALIDATION_PENDING_RETRY:
            // TODO(Ricardo) Send event.
            default:
                on_validation_failed(remote_participant_data, exception);

                std::unique_ptr<DiscoveredParticipantInfo> tmp;
                std::lock_guard<vbs::shared_mutex> _(mutex_);
                {
                    auto it = discovered_participants_.find(participant_data.m_guid);
                    if (it != discovered_participants_.end()) {
                        tmp = std::move(it->second);
                    }

                    // Remove created element, because authentication failed.
                    (void)discovered_participants_.erase(remote_participant_data.m_guid);
                }

                // TODO(Ricardo) cryptograhy registration in AUTHENTICAITON_OK
                return false;
        }

        logDebug(SECURITY, "Discovered participant " << remote_participant_data.m_guid << " " << auth_status);

        // Store new remote handle.
        remote_participant_info->auth_status_ = auth_status;
        remote_participant_info->identity_handle_ = remote_identity_handle;

        // TODO(Ricardo) Start cryptography if authentication ok in this point.
        // If authentication is successful, inform user about it.
        if (auth_status == AUTHENTICATION_OK) {
            // TODO(Ricardo) Shared secret on this case?
            std::shared_ptr<SecretHandle> ss;
            notify_part_authorized = participant_fake_authorized(remote_participant_data, remote_participant_info, ss);
        }
    } else {
        // If cannot retrieve the authentication info pointer then return, because
        // it is used in other thread.
        if (!remote_participant_info) {
            return false;
        }
        if (remote_participant_info->auth_status_ == AUTHENTICATION_FAILED) {
            remote_participant_info->auth_status_ = AUTHENTICATION_REQUEST_NOT_SEND;
        }
    }

    bool returnedValue = true;

    if (remote_participant_info->auth_status_ == AUTHENTICATION_REQUEST_NOT_SEND ||
        remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST) {
        // Maybe send request.
        returnedValue = on_process_fake_handshake(remote_participant_data, remote_participant_info, MessageIdentity(),
                                                  notify_part_authorized);
        if (returnedValue) {
            returnedValue = on_process_fake_handshake(remote_participant_data, remote_participant_info,
                                                      MessageIdentity(), notify_part_authorized);
        }
    }

    (void)restore_discovered_participant_info(remote_participant_data.m_guid, remote_participant_info);
    if (notify_part_authorized) {
        notify_participant_authorized(remote_participant_data);
    }

    return returnedValue;
}

bool SecurityManager::on_process_fake_handshake(const ParticipantProxyData& participant_data,
                                                DiscoveredParticipantInfo::AuthUniquePtr& remote_participant_info,
                                                MessageIdentity&& message_identity, bool& notify_part_authorized) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    HandshakeMessageToken* handshake_message = nullptr;
    SecurityException exception;
    (void)message_identity;

    ValidationResult_t ret = VALIDATION_FAILED;

    assert(remote_participant_info->identity_handle_ != nullptr);

    logDebug(SECURITY, "Processing handshake from participant " << participant_data.m_guid);

    if (remote_participant_info->auth_status_ == AUTHENTICATION_REQUEST_NOT_SEND) {
        ret = authentication_plugin_->begin_handshake_request(
            &remote_participant_info->handshake_handle_, &handshake_message, *local_identity_handle_,
            *remote_participant_info->identity_handle_,
            participant_->pdp()->get_participant_proxy_data_serialized(BIGEND), exception);
    } else if (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REQUEST) {
        assert(!remote_participant_info->handshake_handle_);
        ret = authentication_plugin_->begin_fake_handshake_reply(
            &remote_participant_info->handshake_handle_, *remote_participant_info->identity_handle_,
            *local_identity_handle_, participant_->pdp()->get_participant_proxy_data_serialized(BIGEND), exception);
    } else if ((remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_REPLY) ||
               (remote_participant_info->auth_status_ == AUTHENTICATION_WAITING_FINAL)) {
        assert(remote_participant_info->handshake_handle_);
        auto key = get_shared_key(participant_data.m_guid.guidPrefix);
        if (key) {
            ret = authentication_plugin_->process_fake_handshake(
                remote_participant_info->auth_status_, *key, *remote_participant_info->handshake_handle_, exception);
        }
    } else if (remote_participant_info->auth_status_ == AUTHENTICATION_OK) {
        return true;
    }

    if (ret == VALIDATION_FAILED) {
        remote_participant_info->auth_status_ = AUTHENTICATION_FAILED;
        on_validation_failed(participant_data, exception);
        return false;
    }

    assert(remote_participant_info->handshake_handle_ != nullptr);

    int64_t expected_sequence_number = 0;
    bool handshake_message_send = true;
    bool returnedValue = false;
    AuthenticationStatus pre_auth_status = remote_participant_info->auth_status_;

    if (handshake_message_send) {
        switch (ret) {
            case VALIDATION_OK:
            case VALIDATION_OK_WITH_FINAL_MESSAGE:
            case VALIDATION_PENDING_HANDSHAKE_MESSAGE: {
                remote_participant_info->auth_status_ = AUTHENTICATION_OK;
                if (ret == VALIDATION_PENDING_HANDSHAKE_MESSAGE) {
                    if (pre_auth_status == AUTHENTICATION_REQUEST_NOT_SEND) {
                        remote_participant_info->auth_status_ = AUTHENTICATION_WAITING_REPLY;
                    } else if (pre_auth_status == AUTHENTICATION_WAITING_REQUEST) {
                        remote_participant_info->auth_status_ = AUTHENTICATION_WAITING_FINAL;
                    }
                }

                // if authentication was finished, starts encryption.
                if (remote_participant_info->auth_status_ == AUTHENTICATION_OK) {
                    std::shared_ptr<SecretHandle> shared_secret_handle = authentication_plugin_->get_shared_secret(
                        *remote_participant_info->handshake_handle_, exception);
                    if (!participant_fake_authorized(participant_data, remote_participant_info, shared_secret_handle)) {
                        (void)authentication_plugin_->return_sharedsecret_handle(shared_secret_handle, exception);
                    } else {
                        notify_part_authorized = true;
                    }
                }

                if (ret == VALIDATION_PENDING_HANDSHAKE_MESSAGE) {
                    remote_participant_info->expected_sequence_number_ = expected_sequence_number;
                }

                returnedValue = true;
            } break;
            case VALIDATION_PENDING_RETRY:
            // TODO(Ricardo) Send event.
            default:
                break;
        }
    }

    return returnedValue;
}

void SecurityManager::exchange_fake_participant_crypto(
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto, const GUID_t& remote_participant_guid) {
    SecurityException exception;

    // Get participant crypto tokens.
    ParticipantCryptoTokenSeq local_participant_crypto_tokens;
    if (crypto_plugin_->cryptokeyexchange()->create_fake_participant_crypto_tokens(
            local_participant_crypto_tokens, *local_participant_crypto_handle_, *remote_participant_crypto,
            exception)) {
        (void)remote_participant_guid;
    } else {
        elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Error generating crypto token. (" << exception.what() << ")");
    }
}

bool SecurityManager::participant_fake_authorized(
    const ParticipantProxyData& participant_data,
    const DiscoveredParticipantInfo::AuthUniquePtr& remote_participant_info,
    std::shared_ptr<SecretHandle>& shared_secret_handle) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    logDebug(SECURITY, "Authorized participant " << participant_data.m_guid);

    SecurityException exception;
    PermissionsHandle* remote_permissions = nullptr;

    if (access_plugin_ != nullptr) {
        PermissionsCredentialToken* credential_token = nullptr;
        if (authentication_plugin_->get_authenticated_peer_credential_token(
                &credential_token, *remote_participant_info->identity_handle_, exception)) {
            (void)authentication_plugin_->return_authenticated_peer_credential_token(credential_token, exception);
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                      "Not receive remote permissions of participant " << participant_data.m_guid << " ("
                                                                       << exception.what() << ").");
        }
    }

    if (true || (access_plugin_ == nullptr) || (remote_permissions != nullptr) ||
        get_mvbs_status(*remote_participant_info->identity_handle_)) {
        std::list<std::pair<ReaderProxyData, GUID_t>> temp_readers;
        std::list<std::pair<WriterProxyData, GUID_t>> temp_writers;

        if (crypto_plugin_ != nullptr && !get_mvbs_status(*remote_participant_info->identity_handle_)) {
            // TODO(Ricardo) Study cryptography without sharedsecret
            if (!shared_secret_handle) {
                elogError(SECURITY, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                          "Not shared secret for participant " << participant_data.m_guid);
                return false;
            }

            // Starts cryptography mechanism
            std::shared_ptr<ParticipantCryptoHandle> participant_crypto_handle =
                register_and_match_crypto_endpoint(*remote_participant_info->identity_handle_, *shared_secret_handle);

            // Store cryptography info
            if (participant_crypto_handle && (!participant_crypto_handle->nil())) {
                std::lock_guard<vbs::shared_mutex> _(mutex_);

                // Check there is a pending crypto message.
                auto pending = remote_participant_pending_messages_.find(participant_data.m_guid);

                if (pending != remote_participant_pending_messages_.end()) {
                    if (!crypto_plugin_->cryptokeyexchange()->set_remote_participant_crypto_tokens(
                            *local_participant_crypto_handle_, *participant_crypto_handle, pending->second,
                            exception)) {
                        elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                  "Cannot set remote participant crypto tokens (" << participant_data.m_guid << ") - ("
                                                                                  << exception.what() << ")");
                    }

                    (void)remote_participant_pending_messages_.erase(pending);
                }

                ParticipantCryptoTokenSeq fake_remote_crypto_tokens;
                crypto_plugin_->cryptokeyexchange()->create_fake_participant_crypto_tokens(
                    fake_remote_crypto_tokens, *participant_crypto_handle, *local_participant_crypto_handle_,
                    exception);
                if (!crypto_plugin_->cryptokeyexchange()->set_remote_participant_crypto_tokens(
                        *local_participant_crypto_handle_, *participant_crypto_handle, fake_remote_crypto_tokens,
                        exception)) {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Cannot set remote participant crypto tokens (" << participant_data.m_guid << ") - ("
                                                                              << exception.what() << ")");
                }

                // Search in pendings readers and writers
                auto rit = remote_reader_pending_discovery_messages_.begin();
                while (rit != remote_reader_pending_discovery_messages_.end()) {
                    if (std::get<1>(*rit) == participant_data.m_guid) {
                        temp_readers.push_back(std::make_pair(std::get<0>(*rit), std::get<2>(*rit)));
                        rit = remote_reader_pending_discovery_messages_.erase(rit);
                        continue;
                    }

                    ++rit;
                }

                auto wit = remote_writer_pending_discovery_messages_.begin();
                while (wit != remote_writer_pending_discovery_messages_.end()) {
                    if (std::get<1>(*wit) == participant_data.m_guid) {
                        temp_writers.push_back(std::make_pair(std::get<0>(*wit), std::get<2>(*wit)));
                        wit = remote_writer_pending_discovery_messages_.erase(wit);
                        continue;
                    }

                    ++wit;
                }

                auto dp_it = discovered_participants_.find(participant_data.m_guid);

                if (dp_it != discovered_participants_.end()) {
                    dp_it->second->set_participant_crypto(participant_crypto_handle);
                    dp_it->second->set_shared_secret(shared_secret_handle);
                } else {
                    (void)crypto_plugin_->cryptokeyfactory()->unregister_participant(participant_crypto_handle,
                                                                                     exception);
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Cannot find remote participant " << participant_data.m_guid);
                    return false;
                }
            } else {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Cannot register remote participant in crypto plugin (" << participant_data.m_guid << ")");
                return false;
            }

            exchange_fake_participant_crypto(std::move(participant_crypto_handle), participant_data.m_guid);
        } else {
            {
                vbs::shared_lock<vbs::shared_mutex> _(mutex_);

                // Store shared_secret.
                auto dp_it = discovered_participants_.find(participant_data.m_guid);

                if (dp_it != discovered_participants_.end()) {
                    dp_it->second->set_shared_secret(shared_secret_handle);
                }
            }
            if (!get_mvbs_status(*remote_participant_info->identity_handle_))
                match_builtin_key_exchange_endpoints(participant_data);
        }

        logDebug(SECURITY, "Participant " << participant_data.m_guid << " authenticated");

        if (!get_mvbs_status(*remote_participant_info->identity_handle_)) {
            for (auto& remote_reader : temp_readers) {
                logInfo(SECURITY, "unicastlocator:" << remote_reader.first.remote_locators().unicast.size());
                (void)participant_->pdp()->getEDP()->pairing_reader_proxy_with_local_writer(
                    remote_reader.second, participant_data.m_guid, remote_reader.first);
            }

            for (auto& remote_writer : temp_writers) {
                (void)participant_->pdp()->getEDP()->pairing_writer_proxy_with_local_reader(
                    remote_writer.second, participant_data.m_guid, remote_writer.first);
            }
        }
        return true;
    }

    return false;
}

bool SecurityManager::discovered_writer_without_handshake(const GUID_t& reader_guid,
                                                          const GUID_t& remote_participant_key,
                                                          WriterProxyData& remote_writer_data,
                                                          const EndpointSecurityAttributes& security_attributes,
                                                          bool is_builtin) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    std::unique_lock<vbs::shared_mutex> lock(mutex_);

    PermissionsHandle* remote_permissions = nullptr;
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto_handle;
    std::shared_ptr<SecretHandle> shared_secret_handle;
    AuthenticationStatus auth_status(AUTHENTICATION_INIT);

    if (!security_attributes.match(remote_writer_data.security_attributes_,
                                   remote_writer_data.plugin_security_attributes_)) {
        return false;
    }

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_permissions = dp_it->second->get_permissions_handle();
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
            shared_secret_handle = dp_it->second->get_shared_secret();
            auth_status = dp_it->second->get_auth_status();
            if (is_builtin && dp_it->second->participant_data().is_client) {
                return true;
            }
            if (!dp_it->second->participant_data().is_client) {
                return false;
            }
        }
    }

    // assert(access_plugin_ == nullptr || remote_permissions != nullptr);
    // assert(crypto_plugin_ == nullptr || remote_participant_crypto_handle != nullptr);

    bool returned_value = true;
    SecurityException exception;

    if (!is_builtin) {
        if (participant_->security_attributes().allow_unauthenticated_participants &&
            auth_status != AUTHENTICATION_NOT_AVAILABLE && auth_status != AUTHENTICATION_OK &&
            (security_attributes.is_write_protected || security_attributes.is_read_protected)) {
            //!Do not match if read or write protection is enabled for this local endpoint
            return false;
        }

        if (access_plugin_ != nullptr && remote_permissions != nullptr) {
            if ((returned_value = access_plugin_->check_remote_datawriter(*remote_permissions, domain_id_,
                                                                          remote_writer_data, exception)) == false) {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking create remote writer " << remote_writer_data.guid() << " ("
                                                                 << exception.what() << ")");
            }
        }
    }

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        bool is_key_exchange =
            (remote_writer_data.guid().entityId == participant_volatile_message_secure_writer_entity_id);
        auto local_reader = reader_handles_.find(reader_guid);
        returned_value = false;

        if (local_reader != reader_handles_.end()) {
            if ((remote_participant_crypto_handle != nullptr) && (shared_secret_handle != nullptr)) {
                DatawriterCryptoHandle* remote_writer_handle =
                    crypto_plugin_->cryptokeyfactory()->register_matched_remote_datawriter(
                        *local_reader->second.reader_handle, *remote_participant_crypto_handle, *shared_secret_handle,
                        exception);

                if ((remote_writer_handle != nullptr) && (!remote_writer_handle->nil())) {
                    if (is_key_exchange) {
                        (void)local_reader->second.associated_writers.emplace(
                            remote_writer_data.guid(), std::make_tuple(remote_writer_data, remote_writer_handle));
                        lock.unlock();
                        (void)participant_->pairing_remote_writer_with_local_reader_after_security(reader_guid,
                                                                                                   remote_writer_data);
                    } else {
                        // Check pending writer crypto messages.
                        auto pending = remote_writer_pending_messages_.find(
                            std::make_pair(remote_writer_data.guid(), reader_guid));
                        bool pairing_cause_pending_message = false;

                        if (pending != remote_writer_pending_messages_.end()) {
                            if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                                    *local_reader->second.reader_handle, *remote_writer_handle, pending->second,
                                    exception)) {
                                pairing_cause_pending_message = true;
                            } else {
                                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                          "Cannot set remote writer crypto tokens ("
                                              << remote_writer_data.guid() << ") - (" << exception.what() << ")");
                            }

                            (void)remote_writer_pending_messages_.erase(pending);
                        }

                        GUID_t local_writer_guid;
                        ReaderProxyData* reader_data = nullptr;

                        // Get local reader crypto tokens.
                        DatareaderCryptoTokenSeq local_reader_crypto_tokens;
                        if (crypto_plugin_->cryptokeyexchange()->create_fake_reader_crypto_tokens(
                                local_reader_crypto_tokens, *local_reader->second.reader_handle, *remote_writer_handle,
                                exception)) {
                            if (remote_participant_key == participant_->getGuid()) {
                                logDebug(SECURITY,
                                         "Process successful discovering local writer " << remote_writer_data.guid());
                                (void)local_reader->second.associated_writers.emplace(
                                    remote_writer_data.guid(),
                                    std::make_tuple(remote_writer_data, remote_writer_handle));

                                // Search local writer.
                                auto local_writer = writer_handles_.find(remote_writer_data.guid());

                                if (local_writer != writer_handles_.end()) {
                                    returned_value = true;
                                    auto remote_reader = local_writer->second.associated_readers.find(reader_guid);

                                    if (remote_reader != local_writer->second.associated_readers.end()) {
                                        if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                                                *local_writer->second.writer_handle,
                                                *std::get<1>(remote_reader->second), local_reader_crypto_tokens,
                                                exception)) {
                                            local_writer_guid = local_writer->first;
                                            reader_data = &(std::get<0>(remote_reader->second));
                                        } else {
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "Cannot set local writer crypto tokens ("
                                                          << remote_writer_data.guid() << ") - (" << exception.what()
                                                          << ")");
                                        }
                                    } else {
                                        // Store in pendings.
                                        (void)remote_reader_pending_messages_.emplace(
                                            std::make_pair(reader_guid, local_writer->first),
                                            std::move(local_reader_crypto_tokens));
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot find local writer (" << remote_writer_data.guid() << ") - ("
                                                                           << exception.what() << ")");
                                }
                            } else {
                                auto result = local_reader->second.associated_writers.emplace(
                                    remote_writer_data.guid(),
                                    std::make_tuple(remote_writer_data, remote_writer_handle));
                                if (!result.second) {  // key is exist, update
                                    result.first->second = std::make_tuple(remote_writer_data, remote_writer_handle);
                                }

                                DatawriterCryptoTokenSeq fake_remote_crypto_tokens;
                                crypto_plugin_->cryptokeyexchange()->create_fake_writer_crypto_tokens(
                                    fake_remote_crypto_tokens, *remote_writer_handle,
                                    *local_reader->second.reader_handle, exception);
                                if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                                        *local_reader->second.reader_handle, *remote_writer_handle,
                                        fake_remote_crypto_tokens, exception)) {
                                    returned_value = true;
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot set remote writer crypto tokens ("
                                                  << remote_writer_data.guid() << ") - (" << exception.what() << ")");
                                }
                                lock.unlock();
                            }
                        } else {
                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                      "Error generating crypto token. (" << exception.what() << ")");
                        }

                        // Here the mutex has to be unlock.
                        if (lock) {
                            lock.unlock();
                        }

                        // If writer was found and setting of crypto tokens works,
                        // then tell core to match writer and reader.
                        if (local_writer_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_reader_with_local_writer_after_security(
                                local_writer_guid, *reader_data);
                        }

                        // If reader was found and setting of crypto tokens works,
                        // then tell core to match reader and writer.
                        if (pairing_cause_pending_message || reader_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_writer_with_local_reader_after_security(
                                reader_guid, remote_writer_data);
                        }
                    }
                } else {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Crypto plugin fails registering remote writer "
                                  << remote_writer_data.guid() << " of participant " << remote_participant_key);
                }
            } else {
                logDebug(SECURITY, "Storing remote writer << " << remote_writer_data.guid() << " of participant "
                                                               << remote_participant_key << "on pendings");

                remote_writer_pending_discovery_messages_.push_back(
                    std::make_tuple(remote_writer_data, remote_participant_key, reader_guid));

                returned_value = true;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local reader " << reader_guid << std::endl);
        }
    } else if (returned_value) {
        lock.unlock();
        (void)participant_->pairing_remote_writer_with_local_reader_after_security(reader_guid, remote_writer_data);
    }

    return returned_value;
}
bool SecurityManager::discovered_reader_without_handshake(const GUID_t& writer_guid,
                                                          const GUID_t& remote_participant_key,
                                                          ReaderProxyData& remote_reader_data,
                                                          const EndpointSecurityAttributes& security_attributes,
                                                          bool is_builtin) {
    auto sentry = is_security_manager_initialized();
    if (!sentry) {
        return false;
    }

    std::unique_lock<vbs::shared_mutex> lock(mutex_);

    PermissionsHandle* remote_permissions = nullptr;
    std::shared_ptr<ParticipantCryptoHandle> remote_participant_crypto_handle;
    std::shared_ptr<SecretHandle> shared_secret_handle;
    AuthenticationStatus auth_status(AUTHENTICATION_INIT);

    if (!security_attributes.match(remote_reader_data.security_attributes_,
                                   remote_reader_data.plugin_security_attributes_)) {
        logError(SECURITY,
                 "reader security_attributes unmatch with plugin: " << remote_reader_data.security_attributes_ << " "
                                                                    << remote_reader_data.plugin_security_attributes_);
        return false;
    }

    if (remote_participant_key == participant_->getGuid()) {
        remote_participant_crypto_handle = local_participant_crypto_handle_;
    } else {
        auto dp_it = discovered_participants_.find(remote_participant_key);

        if (dp_it != discovered_participants_.end()) {
            remote_permissions = dp_it->second->get_permissions_handle();
            remote_participant_crypto_handle = dp_it->second->get_participant_crypto();
            shared_secret_handle = dp_it->second->get_shared_secret();
            auth_status = dp_it->second->get_auth_status();
            if (is_builtin && dp_it->second->participant_data().is_client) {
                return true;
            }
            if (!dp_it->second->participant_data().is_client) {
                return false;
            }
        }
    }

    // assert(access_plugin_ == nullptr || remote_permissions != nullptr);
    // assert(crypto_plugin_ == nullptr || remote_participant_crypto_handle != nullptr);

    bool relay_only = false;
    bool returned_value = true;
    SecurityException exception;

    if (!is_builtin) {
        //! Check if it is an unathenticated participant
        if (participant_->security_attributes().allow_unauthenticated_participants &&
            auth_status != AUTHENTICATION_NOT_AVAILABLE && auth_status != AUTHENTICATION_OK &&
            (security_attributes.is_write_protected || security_attributes.is_read_protected)) {
            //!Do not match if read or write protection is enabled for this local endpoint
            return false;
        }

        if (access_plugin_ != nullptr && remote_permissions != nullptr) {
            if ((returned_value = access_plugin_->check_remote_datareader(
                     *remote_permissions, domain_id_, remote_reader_data, relay_only, exception)) == false) {
                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                          "Error checking create remote reader " << remote_reader_data.guid() << " ("
                                                                 << exception.what() << ")");
            }
        }
    }

    if (returned_value && (crypto_plugin_ != nullptr) &&
        (security_attributes.is_submessage_protected || security_attributes.is_payload_protected)) {
        bool is_key_exchange =
            (remote_reader_data.guid().entityId == participant_volatile_message_secure_reader_entity_id);
        auto local_writer = writer_handles_.find(writer_guid);
        returned_value = false;

        if (local_writer != writer_handles_.end()) {
            if (remote_participant_crypto_handle != nullptr && shared_secret_handle != nullptr) {
                DatareaderCryptoHandle* remote_reader_handle =
                    crypto_plugin_->cryptokeyfactory()->register_matched_remote_datareader(
                        *local_writer->second.writer_handle, *remote_participant_crypto_handle, *shared_secret_handle,
                        relay_only, exception);

                if ((remote_reader_handle != nullptr) && (!remote_reader_handle->nil())) {
                    if (is_key_exchange) {
                        logDebug(SECURITY, "Process successful discovering local reader "
                                               << remote_reader_data.guid() << " unicastlocatorsize:"
                                               << remote_reader_data.remote_locators().unicast.size());
                        (void)local_writer->second.associated_readers.emplace(
                            remote_reader_data.guid(), std::make_tuple(remote_reader_data, remote_reader_handle));
                        lock.unlock();
                        (void)participant_->pairing_remote_reader_with_local_writer_after_security(writer_guid,
                                                                                                   remote_reader_data);
                    } else {
                        // Check pending reader crypto messages.
                        auto pending = remote_reader_pending_messages_.find(
                            std::make_pair(remote_reader_data.guid(), writer_guid));
                        bool pairing_cause_pending_message = false;

                        if (pending != remote_reader_pending_messages_.end()) {
                            if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                                    *local_writer->second.writer_handle, *remote_reader_handle, pending->second,
                                    exception)) {
                                pairing_cause_pending_message = true;
                            } else {
                                elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                          "Cannot set remote reader crypto tokens ("
                                              << remote_reader_data.guid() << ") - (" << exception.what() << ")");
                            }

                            (void)remote_reader_pending_messages_.erase(pending);
                        }

                        GUID_t local_reader_guid;
                        WriterProxyData* writer_data = nullptr;

                        // Get local writer crypto tokens.
                        DatawriterCryptoTokenSeq local_writer_crypto_tokens;
                        if (crypto_plugin_->cryptokeyexchange()->create_fake_writer_crypto_tokens(
                                local_writer_crypto_tokens, *local_writer->second.writer_handle, *remote_reader_handle,
                                exception)) {
                            if (remote_participant_key == participant_->getGuid()) {
                                logDebug(SECURITY, "Process successful discovering local reader "
                                                       << remote_reader_data.guid() << " unicastlocatorsize:"
                                                       << remote_reader_data.remote_locators().unicast.size());
                                (void)local_writer->second.associated_readers.emplace(
                                    remote_reader_data.guid(),
                                    std::make_tuple(remote_reader_data, remote_reader_handle));

                                // Search local reader.
                                auto local_reader = reader_handles_.find(remote_reader_data.guid());

                                if (local_reader != reader_handles_.end()) {
                                    returned_value = true;
                                    auto remote_writer = local_reader->second.associated_writers.find(writer_guid);

                                    if (remote_writer != local_reader->second.associated_writers.end()) {
                                        if (crypto_plugin_->cryptokeyexchange()->set_remote_datawriter_crypto_tokens(
                                                *local_reader->second.reader_handle,
                                                *std::get<1>(remote_writer->second), local_writer_crypto_tokens,
                                                exception)) {
                                            local_reader_guid = local_reader->first;
                                            writer_data = &(std::get<0>(remote_writer->second));
                                        } else {
                                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                                      "Cannot set local reader crypto tokens ("
                                                          << remote_reader_data.guid() << ") - (" << exception.what()
                                                          << ")");
                                        }
                                    } else {
                                        // Store in pendings.
                                        (void)remote_writer_pending_messages_.emplace(
                                            std::make_pair(writer_guid, local_reader->first),
                                            std::move(local_writer_crypto_tokens));
                                    }
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot find local reader (" << remote_reader_data.guid() << ") - ("
                                                                           << exception.what() << ")");
                                }
                            } else {
                                auto result = local_writer->second.associated_readers.emplace(
                                    remote_reader_data.guid(),
                                    std::make_tuple(remote_reader_data, remote_reader_handle));
                                if (!result.second) {  // key is exist, update
                                    result.first->second = std::make_tuple(remote_reader_data, remote_reader_handle);
                                }

                                DatareaderCryptoTokenSeq fake_remote_crypto_tokens;
                                crypto_plugin_->cryptokeyexchange()->create_fake_reader_crypto_tokens(
                                    fake_remote_crypto_tokens, *remote_reader_handle,
                                    *local_writer->second.writer_handle, exception);
                                if (crypto_plugin_->cryptokeyexchange()->set_remote_datareader_crypto_tokens(
                                        *local_writer->second.writer_handle, *remote_reader_handle,
                                        fake_remote_crypto_tokens, exception)) {
                                    returned_value = true;
                                } else {
                                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                              "Cannot set remote writer crypto tokens ("
                                                  << remote_reader_data.guid() << ") - (" << exception.what() << ")");
                                }
                                lock.unlock();
                            }
                        } else {
                            elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                                      "Error generating crypto token. (" << exception.what() << ")");
                        }

                        // Here the mutex has to be unlock.
                        if (lock) {
                            lock.unlock();
                        }

                        // If reader was found and setting of crypto tokens works,
                        // then tell core to match reader and writer.
                        if (local_reader_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_writer_with_local_reader_after_security(
                                local_reader_guid, *writer_data);
                        }

                        // If writer was found and setting of crypto tokens works,
                        // then tell core to match writer and reader.
                        if (pairing_cause_pending_message || writer_guid != GUID_t::unknown()) {
                            (void)participant_->pairing_remote_reader_with_local_writer_after_security(
                                writer_guid, remote_reader_data);
                        }
                    }
                } else {
                    elogError(SECURITY, RetCode_t::RETCODE_ERROR,
                              "Crypto plugin fails registering remote reader "
                                  << remote_reader_data.guid() << " of participant " << remote_participant_key);
                }
            } else {
                logDebug(SECURITY, "Storing remote reader << " << remote_reader_data.guid() << " of participant "
                                                               << remote_participant_key << " on pendings");

                remote_reader_pending_discovery_messages_.push_back(
                    std::make_tuple(remote_reader_data, remote_participant_key, writer_guid));

                returned_value = true;
            }
        } else {
            elogError(SECURITY, RetCode_t::RETCODE_ERROR, "Cannot find local writer " << writer_guid << std::endl);
        }
    } else if (returned_value) {
        lock.unlock();
        (void)participant_->pairing_remote_reader_with_local_writer_after_security(writer_guid, remote_reader_data);
    }

    return returned_value;
}

const std::vector<uint8_t>* SecurityManager::get_shared_key(const GuidPrefix_t& peer) const {
    if (crypto_plugin_ != nullptr) {
        return crypto_plugin_->cryptokeyfactory()->get_shared_key(peer);
    } else {
        auto it = shared_keys_.find(peer);
        if (it != shared_keys_.end()) {
            return &it->second;
        }
        return nullptr;
    }
}

void SecurityManager::set_shared_key(const GuidPrefix_t& peer, const std::vector<uint8_t>& key) {
    if (key.size() == 0) {
        logInfo(RTPS_MSG_IN, "shared_key size error:" << key.size());
        return;
    }

    if (crypto_plugin_ != nullptr) {
        crypto_plugin_->cryptokeyfactory()->set_shared_key(peer, key);
    } else {
        shared_keys_[peer] = key;
        logDebug(SECURITY, "add shared key:" << peer);
    }
}

}  // namespace rtps
}  // namespace ertps
}  // namespace evbs
